diff --git a/iota/exceptions.py b/iota/exceptions.py index eaa71e1..31129c7 100644 --- a/iota/exceptions.py +++ b/iota/exceptions.py @@ -1,25 +1,36 @@ # coding=utf-8 from __future__ import absolute_import, division, print_function, \ - unicode_literals + unicode_literals + +__all__ = [ + 'with_context', +] def with_context(exc, context): - # type: (Exception, dict) -> Exception - """ - Attaches a ``context`` value to an Exception. + # type: (Exception, dict) -> Exception + """ + Attaches a ``context`` value to an Exception. + + Before: + + .. code-block:: python - Before:: + exc = Exception('Frog blast the vent core!') + exc.context = { ... } + raise exc - exc = Exception('Frog blast the vent core!') - exc.context = { ... } - raise exc + After: - After:: + .. code-block:: python - raise with_context(Exception('Frog blast the vent core!'), { ... }) - """ - if not hasattr(exc, 'context'): - exc.context = {} + raise with_context( + exc=Exception('Frog blast the vent core!'), + context={ ... }, + ) + """ + if not hasattr(exc, 'context'): + exc.context = {} - exc.context.update(context) - return exc + exc.context.update(context) + return exc diff --git a/iota/filters.py b/iota/filters.py index 1c1ffdb..2c98067 100644 --- a/iota/filters.py +++ b/iota/filters.py @@ -1,6 +1,6 @@ # coding=utf-8 from __future__ import absolute_import, division, print_function, \ - unicode_literals + unicode_literals from typing import Text @@ -13,192 +13,201 @@ class GeneratedAddress(f.BaseFilter): - """ - Validates an incoming value as a generated :py:class:`Address` (must - have ``key_index`` set). - """ - CODE_NO_KEY_INDEX = 'no_key_index' - CODE_NO_SECURITY_LEVEL = 'no_security_level' + """ + Validates an incoming value as a generated :py:class:`Address` (must + have ``key_index`` set). + """ + CODE_NO_KEY_INDEX = 'no_key_index' + CODE_NO_SECURITY_LEVEL = 'no_security_level' - templates = { - CODE_NO_KEY_INDEX: - 'Address must have ``key_index`` attribute set.', + templates = { + CODE_NO_KEY_INDEX: + 'Address must have ``key_index`` attribute set.', - CODE_NO_SECURITY_LEVEL: - 'Address must have ``security_level`` attribute set.', - } + CODE_NO_SECURITY_LEVEL: + 'Address must have ``security_level`` attribute set.', + } - def _apply(self, value): - value = self._filter(value, f.Type(Address)) # type: Address + def _apply(self, value): + value = self._filter(value, f.Type(Address)) # type: Address - if self._has_errors: - return None + if self._has_errors: + return None - if value.key_index is None: - return self._invalid_value(value, self.CODE_NO_KEY_INDEX) + if value.key_index is None: + return self._invalid_value(value, self.CODE_NO_KEY_INDEX) - if value.security_level is None: - return self._invalid_value(value, self.CODE_NO_SECURITY_LEVEL) + if value.security_level is None: + return self._invalid_value(value, self.CODE_NO_SECURITY_LEVEL) - return value + return value class NodeUri(f.BaseFilter): - """ - Validates a string as a node URI. - """ - SCHEMES = {'tcp', 'udp'} - """ - Allowed schemes for node URIs. - """ + """ + Validates a string as a node URI. + """ + SCHEMES = {'tcp', 'udp'} + """ + Allowed schemes for node URIs. + """ - CODE_NOT_NODE_URI = 'not_node_uri' + CODE_NOT_NODE_URI = 'not_node_uri' - templates = { - CODE_NOT_NODE_URI: 'This value does not appear to be a valid node URI.', - } + templates = { + CODE_NOT_NODE_URI: + 'This value does not appear to be a valid node URI.', + } - def _apply(self, value): - value = self._filter(value, f.Type(text_type)) # type: Text + def _apply(self, value): + value = self._filter(value, f.Type(text_type)) # type: Text - if self._has_errors: - return None + if self._has_errors: + return None - parsed = compat.urllib_parse.urlparse(value) + parsed = compat.urllib_parse.urlparse(value) - if parsed.scheme not in self.SCHEMES: - return self._invalid_value(value, self.CODE_NOT_NODE_URI) + if parsed.scheme not in self.SCHEMES: + return self._invalid_value(value, self.CODE_NOT_NODE_URI) - return value + return value +# noinspection PyPep8Naming @filter_macro def SecurityLevel(): - """ - Generates a filter chain for validating a security level. - """ - return ( - f.Type(int) | - f.Min(1) | - f.Max(3) | - f.Optional(default=AddressGenerator.DEFAULT_SECURITY_LEVEL) - ) + """ + Generates a filter chain for validating a security level. + """ + return ( + f.Type(int) | + f.Min(1) | + f.Max(3) | + f.Optional(default=AddressGenerator.DEFAULT_SECURITY_LEVEL) + ) class Trytes(f.BaseFilter): - """ - Validates a sequence as a sequence of trytes. - """ - CODE_NOT_TRYTES = 'not_trytes' - CODE_WRONG_FORMAT = 'wrong_format' - - templates = { - CODE_NOT_TRYTES: 'This value is not a valid tryte sequence.', - CODE_WRONG_FORMAT: 'This value is not a valid {result_type}.', - } - - def __init__(self, result_type=TryteString): - # type: (type) -> None - super(Trytes, self).__init__() - - if not isinstance(result_type, type): - raise TypeError( - 'Invalid result_type for {filter_type} ' - '(expected subclass of TryteString, ' - 'actual instance of {result_type}).'.format( - filter_type = type(self).__name__, - result_type = type(result_type).__name__, - ), - ) - - if not issubclass(result_type, TryteString): - raise ValueError( - 'Invalid result_type for {filter_type} ' - '(expected TryteString, actual {result_type}).'.format( - filter_type = type(self).__name__, - result_type = result_type.__name__, - ), - ) - - self.result_type = result_type - - def _apply(self, value): - # noinspection PyTypeChecker - value =\ - self._filter( - filter_chain = f.Type((binary_type, bytearray, text_type, TryteString)), - value = value, - ) # type: TrytesCompatible - - if self._has_errors: - return None - - # If the incoming value already has the correct type, then we're - # done. - if isinstance(value, self.result_type): - return value - - # First convert to a generic TryteString, to make sure that the - # sequence doesn't contain any invalid characters. - try: - value = TryteString(value) - except ValueError: - return self._invalid_value(value, self.CODE_NOT_TRYTES, exc_info=True) - - if self.result_type is TryteString: - return value - - # Now coerce to the expected type and verify that there are no - # type-specific errors. - try: - return self.result_type(value) - except ValueError: - return self._invalid_value( - value = value, - reason = self.CODE_WRONG_FORMAT, - exc_info = True, - - template_vars = { - 'result_type': self.result_type.__name__, - }, - ) + """ + Validates a sequence as a sequence of trytes. + """ + CODE_NOT_TRYTES = 'not_trytes' + CODE_WRONG_FORMAT = 'wrong_format' + + templates = { + CODE_NOT_TRYTES: 'This value is not a valid tryte sequence.', + CODE_WRONG_FORMAT: 'This value is not a valid {result_type}.', + } + + def __init__(self, result_type=TryteString): + # type: (type) -> None + super(Trytes, self).__init__() + + if not isinstance(result_type, type): + raise TypeError( + 'Invalid result_type for {filter_type} ' + '(expected subclass of TryteString, ' + 'actual instance of {result_type}).'.format( + filter_type=type(self).__name__, + result_type=type(result_type).__name__, + ), + ) + + if not issubclass(result_type, TryteString): + raise ValueError( + 'Invalid result_type for {filter_type} ' + '(expected TryteString, actual {result_type}).'.format( + filter_type=type(self).__name__, + result_type=result_type.__name__, + ), + ) + + self.result_type = result_type + + def _apply(self, value): + # noinspection PyTypeChecker + value = self._filter( + filter_chain=f.Type( + (binary_type, bytearray, text_type, TryteString) + ), + + value=value, + ) # type: TrytesCompatible + + if self._has_errors: + return None + + # If the incoming value already has the correct type, then we're + # done. + if isinstance(value, self.result_type): + return value + + # First convert to a generic TryteString, to make sure that the + # sequence doesn't contain any invalid characters. + try: + value = TryteString(value) + except ValueError: + return self._invalid_value( + value=value, + reason=self.CODE_NOT_TRYTES, + exc_info=True, + ) + + if self.result_type is TryteString: + return value + + # Now coerce to the expected type and verify that there are no + # type-specific errors. + try: + return self.result_type(value) + except ValueError: + return self._invalid_value( + value=value, + reason=self.CODE_WRONG_FORMAT, + exc_info=True, + + template_vars={ + 'result_type': self.result_type.__name__, + }, + ) class AddressNoChecksum(Trytes): - """ - Validates a sequence as an Address then chops off the checksum if it exists - """ - ADDRESS_BAD_CHECKSUM = 'address_bad_checksum' - - templates = { - ADDRESS_BAD_CHECKSUM: - 'Checksum is {supplied_checksum}, should be {expected_checksum}?', - } - - def __init__(self): - super(AddressNoChecksum, self).__init__(result_type=Address) - - def _apply(self, value): - super(AddressNoChecksum, self)._apply(value) - - if self._has_errors: - return None - - # Possible it's still just a TryteString - if not isinstance(value, Address): - value = Address(value) - - # Bail out if we have a bad checksum - if value.checksum and not value.is_checksum_valid(): - return self._invalid_value( - value = value, - reason = self.ADDRESS_BAD_CHECKSUM, - exc_info = True, - - context = { - 'supplied_checksum': value.checksum, - 'expected_checksum': value.with_valid_checksum().checksum, - }, - ) - - return Address(value.address) + """ + Validates a sequence as an Address, then chops off the checksum if + present. + """ + ADDRESS_BAD_CHECKSUM = 'address_bad_checksum' + + templates = { + ADDRESS_BAD_CHECKSUM: + 'Checksum is {supplied_checksum}, should be {expected_checksum}?', + } + + def __init__(self): + super(AddressNoChecksum, self).__init__(result_type=Address) + + def _apply(self, value): + super(AddressNoChecksum, self)._apply(value) + + if self._has_errors: + return None + + # Possible it's still just a TryteString. + if not isinstance(value, Address): + value = Address(value) + + # Bail out if we have a bad checksum. + if value.checksum and not value.is_checksum_valid(): + return self._invalid_value( + value=value, + reason=self.ADDRESS_BAD_CHECKSUM, + exc_info=True, + + context={ + 'supplied_checksum': value.checksum, + 'expected_checksum': value.with_valid_checksum().checksum, + }, + ) + + return Address(value.address) diff --git a/iota/json.py b/iota/json.py index eab810b..61e6c7f 100644 --- a/iota/json.py +++ b/iota/json.py @@ -1,67 +1,75 @@ # coding=utf-8 from __future__ import absolute_import, division, print_function, \ - unicode_literals + unicode_literals from abc import ABCMeta, abstractmethod as abstract_method from json.encoder import JSONEncoder as BaseJsonEncoder -from typing import Iterable, Mapping, Text +from typing import Iterable, Mapping from six import with_metaclass class JsonSerializable(with_metaclass(ABCMeta)): - """ - Interface for classes that can be safely converted to JSON. - """ - @abstract_method - def as_json_compatible(self): """ - Returns a JSON-compatible representation of the object. - - References: - - :py:class:`iota.json.JsonEncoder`. + Interface for classes that can be safely converted to JSON. """ - raise NotImplementedError( - 'Not implemented in {cls}.'.format(cls=type(self).__name__), - ) - def _repr_pretty_(self, p, cycle): - """ - Makes JSON-serializable objects play nice with IPython's default - pretty-printer. + @abstract_method + def as_json_compatible(self): + """ + Returns a JSON-compatible representation of the object. - Sadly, :py:func:`pprint.pprint` does not have a similar mechanism. + References: - References: - - http://ipython.readthedocs.io/en/stable/api/generated/IPython.lib.pretty.html - - :py:meth:`IPython.lib.pretty.RepresentationPrinter.pretty` - - :py:func:`pprint._safe_repr` - """ - # type: (JsonSerializable, bool) -> Text - class_name = type(self).__name__ + - :py:class:`iota.json.JsonEncoder`. + """ + raise NotImplementedError( + 'Not implemented in {cls}.'.format(cls=type(self).__name__), + ) + + def _repr_pretty_(self, p, cycle): + """ + Makes JSON-serializable objects play nice with IPython's default + pretty-printer. + + Sadly, :py:func:`pprint.pprint` does not have a similar + mechanism. + + References: - if cycle: - p.text('{cls}(...)'.format( - cls = class_name, - )) - else: - with p.group(len(class_name)+1, '{cls}('.format(cls=class_name), ')'): - prepared = self.as_json_compatible() - if isinstance(prepared, Mapping): - p.text('**') - elif isinstance(prepared, Iterable): - p.text('*') + - http://ipython.readthedocs.io/en/stable/api/generated/IPython.lib.pretty.html + - :py:meth:`IPython.lib.pretty.RepresentationPrinter.pretty` + - :py:func:`pprint._safe_repr` + """ + class_name = type(self).__name__ - p.pretty(prepared) + if cycle: + p.text('{cls}(...)'.format( + cls=class_name, + )) + else: + with p.group( + len(class_name) + 1, + '{cls}('.format(cls=class_name), + ')', + ): + prepared = self.as_json_compatible() + + if isinstance(prepared, Mapping): + p.text('**') + elif isinstance(prepared, Iterable): + p.text('*') + + p.pretty(prepared) -# noinspection PyClassHasNoInit class JsonEncoder(BaseJsonEncoder): - """ - JSON encoder with support for :py:class:`JsonSerializable`. - """ - def default(self, o): - if isinstance(o, JsonSerializable): - return o.as_json_compatible() - - return super(JsonEncoder, self).default(o) + """ + JSON encoder with support for :py:class:`JsonSerializable`. + """ + + def default(self, o): + if isinstance(o, JsonSerializable): + return o.as_json_compatible() + + return super(JsonEncoder, self).default(o) diff --git a/iota/trits.py b/iota/trits.py index 872ad92..189af50 100644 --- a/iota/trits.py +++ b/iota/trits.py @@ -2,127 +2,131 @@ """ Provides functions for manipulating sequences of trits. -Based on: +Based on https://github.com/iotaledger/iota.lib.js/blob/v0.4.2/lib/crypto/helpers/adder.js """ from __future__ import absolute_import, division, print_function, \ - unicode_literals + unicode_literals from typing import Iterable, List, Optional, Sequence, Tuple __all__ = [ - 'add_trits', - 'int_from_trits', - 'trits_from_int', + 'add_trits', + 'int_from_trits', + 'trits_from_int', ] def add_trits(left, right): - # type: (Sequence[int], Sequence[int]) -> List[int] - """ - Adds two sequences of trits together. + # type: (Sequence[int], Sequence[int]) -> List[int] + """ + Adds two sequences of trits together. - The result is a list of trits equal in length to the longer of the - two sequences. + The result is a list of trits equal in length to the longer of the + two sequences. - Note: Overflow is possible. - For example, ``add_trits([1], [1])`` returns ``[-1]``. - """ - target_len = max(len(left), len(right)) + .. note:: + Overflow is possible. - res = [0] * target_len - left += [0] * (target_len - len(left)) - right += [0] * (target_len - len(right)) + For example, ``add_trits([1], [1])`` returns ``[-1]``. + """ + target_len = max(len(left), len(right)) - carry = 0 - for i in range(len(res)): - res[i], carry = _full_add_trits(left[i], right[i], carry) + res = [0] * target_len + left += [0] * (target_len - len(left)) + right += [0] * (target_len - len(right)) - return res + carry = 0 + for i in range(len(res)): + res[i], carry = _full_add_trits(left[i], right[i], carry) + + return res def int_from_trits(trits): - # type: (Iterable[int]) -> int - """ - Converts a sequence of trits into an integer value. - """ - # Normally we'd have to wrap ``enumerate`` inside ``reversed``, but - # balanced ternary puts least significant digits first. - return sum(base * (3 ** power) for power, base in enumerate(trits)) + # type: (Iterable[int]) -> int + """ + Converts a sequence of trits into an integer value. + """ + # Normally we'd have to wrap ``enumerate`` inside ``reversed``, but + # balanced ternary puts least significant digits first. + return sum(base * (3 ** power) for power, base in enumerate(trits)) def trits_from_int(n, pad=1): - # type: (int, Optional[int]) -> List[int] - """ - Returns a trit representation of an integer value. + # type: (int, Optional[int]) -> List[int] + """ + Returns a trit representation of an integer value. + + :param n: + Integer value to convert. - :param n: - Integer value to convert. + :param pad: + Ensure the result has at least this many trits. - :param pad: - Ensure the result has at least this many trits. + References: - References: - https://dev.to/buntine/the-balanced-ternary-machines-of-soviet-russia - https://en.wikipedia.org/wiki/Balanced_ternary - https://rosettacode.org/wiki/Balanced_ternary#Python - """ - if n == 0: - trits = [] - else: - quotient, remainder = divmod(n, 3) + """ + if n == 0: + trits = [] + else: + quotient, remainder = divmod(n, 3) - if remainder == 2: - # Lend 1 to the next place so we can make this trit negative. - quotient += 1 - remainder = -1 + if remainder == 2: + # Lend 1 to the next place so we can make this trit + # negative. + quotient += 1 + remainder = -1 - trits = [remainder] + trits_from_int(quotient, pad=0) + trits = [remainder] + trits_from_int(quotient, pad=0) - if pad: - trits += [0] * max(0, pad - len(trits)) + if pad: + trits += [0] * max(0, pad - len(trits)) - return trits + return trits def _cons_trits(left, right): - # type: (int, int) -> int - """ - Compares two trits. If they have the same value, returns that value. - Otherwise, returns 0. - """ - return left if left == right else 0 + # type: (int, int) -> int + """ + Compares two trits. If they have the same value, returns that + value. Otherwise, returns 0. + """ + return left if left == right else 0 def _add_trits(left, right): - # type: (int, int) -> int - """ - Adds two individual trits together. + # type: (int, int) -> int + """ + Adds two individual trits together. - The result is always a single trit. - """ - res = left + right - return res if -2 < res < 2 else (res < 0) - (res > 0) + The result is always a single trit. + """ + res = left + right + return res if -2 < res < 2 else (res < 0) - (res > 0) def _any_trits(left, right): - # type: (int, int) -> int - """ - Adds two individual trits together and returns a single trit - indicating whether the result is positive or negative. - """ - res = left + right - return (res > 0) - (res < 0) + # type: (int, int) -> int + """ + Adds two individual trits together and returns a single trit + indicating whether the result is positive or negative. + """ + res = left + right + return (res > 0) - (res < 0) def _full_add_trits(left, right, carry): - # type: (int, int, int) -> Tuple[int, int] - """ - Adds two trits together, with support for a carry trit. - """ - sum_both = _add_trits(left, right) - cons_left = _cons_trits(left, right) - cons_right = _cons_trits(sum_both, carry) - - return _add_trits(sum_both, carry), _any_trits(cons_left, cons_right) + # type: (int, int, int) -> Tuple[int, int] + """ + Adds two trits together, with support for a carry trit. + """ + sum_both = _add_trits(left, right) + cons_left = _cons_trits(left, right) + cons_right = _cons_trits(sum_both, carry) + + return _add_trits(sum_both, carry), _any_trits(cons_left, cons_right) diff --git a/iota/types.py b/iota/types.py index 609d1d8..6ebba80 100644 --- a/iota/types.py +++ b/iota/types.py @@ -1,17 +1,17 @@ # coding=utf-8 from __future__ import absolute_import, division, print_function, \ - unicode_literals + unicode_literals from codecs import decode, encode from itertools import chain from math import ceil from random import SystemRandom from typing import Any, AnyStr, Generator, Iterable, Iterator, List, \ - MutableSequence, Optional, Text, Type, TypeVar, Union + MutableSequence, Optional, Text, Type, TypeVar, Union from warnings import warn from six import PY2, binary_type, itervalues, python_2_unicode_compatible, \ - text_type + text_type from iota import AsciiTrytesCodec, TRITS_PER_TRYTE from iota.crypto import HASH_LENGTH @@ -21,845 +21,899 @@ from iota.trits import int_from_trits, trits_from_int __all__ = [ - 'Address', - 'AddressChecksum', - 'Hash', - 'Tag', - 'TrytesCompatible', - 'TryteString', + 'Address', + 'AddressChecksum', + 'Hash', + 'Tag', + 'TrytesCompatible', + 'TryteString', ] - # Custom types for type hints and docstrings. TrytesCompatible = Union[AnyStr, bytearray, 'TryteString'] - T = TypeVar('T', bound='TryteString') + @python_2_unicode_compatible class TryteString(JsonSerializable): - """ - A string representation of a sequence of trytes. - - A tryte string is similar in concept to Python's byte string, except - it has a more limited alphabet. Byte strings are limited to ASCII - (256 possible values), while the tryte string alphabet only has 27 - characters (one for each possible tryte configuration). - - IMPORTANT: A TryteString does not represent a numeric value! - """ - @classmethod - def random(cls, length): - # type: (int) -> TryteString - """ - Generates a random sequence of trytes. - - :param length: - Number of trytes to generate. - """ - alphabet = list(itervalues(AsciiTrytesCodec.alphabet)) - generator = SystemRandom() - - # :py:meth:`SystemRandom.choices` wasn't added until Python 3.6; - # for compatibility, we will continue to use ``choice`` in a loop. - # https://docs.python.org/3/library/random.html#random.choices - return cls( - ''.join(chr(generator.choice(alphabet)) for _ in range(length)) - .encode('ascii') - ) - - @classmethod - def from_bytes(cls, bytes_, codec=AsciiTrytesCodec.name, *args, **kwargs): - # type: (Type[T], Union[binary_type, bytearray], Text, *Any, **Any) -> T - """ - Creates a TryteString from a sequence of bytes. - - :param bytes_: - Source bytes. - - :param codec: - Reserved for future use. - - See https://github.com/iotaledger/iota.lib.py/issues/62 for more - information. - - :param args: - Additional positional arguments to pass to the initializer. - - :param kwargs: - Additional keyword arguments to pass to the initializer. """ - return cls(encode(bytes_, codec), *args, **kwargs) + A string representation of a sequence of trytes. + + A tryte string is similar in concept to Python's byte string, except + it has a more limited alphabet. Byte strings are limited to ASCII + (256 possible values), while the tryte string alphabet only has 27 + characters (one for each possible tryte configuration). + + .. important:: + A TryteString does not represent a numeric value! + """ + + @classmethod + def random(cls, length): + # type: (int) -> TryteString + """ + Generates a random sequence of trytes. + + :param length: + Number of trytes to generate. + """ + alphabet = list(itervalues(AsciiTrytesCodec.alphabet)) + generator = SystemRandom() + + # :py:meth:`SystemRandom.choices` wasn't added until Python 3.6; + # for compatibility, we will continue to use ``choice`` in a + # loop. + # https://docs.python.org/3/library/random.html#random.choices + return cls( + ''.join( + chr(generator.choice(alphabet)) for _ in range(length) + ).encode('ascii') + ) - @classmethod - def from_unicode(cls, string, *args, **kwargs): - # type: (Type[T], Text, *Any, **Any) -> T - """ - Creates a TryteString from a Unicode string. + @classmethod + def from_bytes(cls, bytes_, codec=AsciiTrytesCodec.name, *args, **kwargs): + # type: (Type[T], Union[binary_type, bytearray], Text, *Any, **Any) -> T + """ + Creates a TryteString from a sequence of bytes. + + :param bytes_: + Source bytes. + + :param codec: + Reserved for future use. + + See https://github.com/iotaledger/iota.lib.py/issues/62 for + more information. + + :param args: + Additional positional arguments to pass to the initializer. + + :param kwargs: + Additional keyword arguments to pass to the initializer. + """ + return cls(encode(bytes_, codec), *args, **kwargs) + + @classmethod + def from_unicode(cls, string, *args, **kwargs): + # type: (Type[T], Text, *Any, **Any) -> T + """ + Creates a TryteString from a Unicode string. + + :param string: + Source string. + + :param args: + Additional positional arguments to pass to the initializer. + + :param kwargs: + Additional keyword arguments to pass to the initializer. + """ + return cls.from_bytes( + bytes_=string.encode('utf-8'), + codec=AsciiTrytesCodec.name, + *args, + **kwargs + ) - :param string: - Source string. + @classmethod + def from_string(cls, *args, **kwargs): + """ + Deprecated; use :py:meth:`from_unicode` instead. - :param args: - Additional positional arguments to pass to the initializer. + https://github.com/iotaledger/iota.lib.py/issues/90 + """ + warn( + category=DeprecationWarning, - :param kwargs: - Additional keyword arguments to pass to the initializer. - """ - return cls.from_bytes( - bytes_ = string.encode('utf-8'), - codec = AsciiTrytesCodec.name, - *args, - **kwargs - ) - - @classmethod - def from_string(cls, *args, **kwargs): - """ - Deprecated; use :py:meth:`from_unicode` instead. - - https://github.com/iotaledger/iota.lib.py/issues/90 - """ - warn( - message='`from_string()` is deprecated; use `from_unicode()` instead.', - category=DeprecationWarning, - ) - return cls.from_unicode(*args, **kwargs) - - @classmethod - def from_trytes(cls, trytes, *args, **kwargs): - # type: (Type[T], Iterable[Iterable[int]], *Any, **Any) -> T - """ - Creates a TryteString from a sequence of trytes. + message=( + '`from_string()` is deprecated; use `from_unicode()` instead.' + ), + ) + return cls.from_unicode(*args, **kwargs) - :param trytes: - Iterable of tryte values. - In this context, a tryte is defined as a list containing 3 trits. + @classmethod + def from_trytes(cls, trytes, *args, **kwargs): + # type: (Type[T], Iterable[Iterable[int]], *Any, **Any) -> T + """ + Creates a TryteString from a sequence of trytes. - :param args: - Additional positional arguments to pass to the initializer. + :param trytes: + Iterable of tryte values. - :param kwargs: - Additional keyword arguments to pass to the initializer. + In this context, a tryte is defined as a list containing 3 + trits. - References: - - :py:meth:`as_trytes` - """ - chars = bytearray() + :param args: + Additional positional arguments to pass to the initializer. - for t in trytes: - converted = int_from_trits(t) + :param kwargs: + Additional keyword arguments to pass to the initializer. - # :py:meth:`_tryte_from_int` - if converted < 0: - converted += 27 + References: - chars.append(AsciiTrytesCodec.alphabet[converted]) + - :py:meth:`as_trytes` + """ + chars = bytearray() - return cls(chars, *args, **kwargs) + for t in trytes: + converted = int_from_trits(t) - @classmethod - def from_trits(cls, trits, *args, **kwargs): - # type: (Type[T], Iterable[int], *Any, **Any) -> T - """ - Creates a TryteString from a sequence of trits. + # :py:meth:`_tryte_from_int` + if converted < 0: + converted += 27 - :param trits: - Iterable of trit values (-1, 0, 1). + chars.append(AsciiTrytesCodec.alphabet[converted]) - :param args: - Additional positional arguments to pass to the initializer. + return cls(chars, *args, **kwargs) - :param kwargs: - Additional keyword arguments to pass to the initializer. + @classmethod + def from_trits(cls, trits, *args, **kwargs): + # type: (Type[T], Iterable[int], *Any, **Any) -> T + """ + Creates a TryteString from a sequence of trits. - References: - - :py:func:`int_from_trits` - - :py:meth:`as_trits` - """ - # Allow passing a generator or other non-Sized value to this - # method. - trits = list(trits) + :param trits: + Iterable of trit values (-1, 0, 1). - if len(trits) % 3: - # Pad the trits so that it is cleanly divisible into trytes. - trits += [0] * (3 - (len(trits) % 3)) + :param args: + Additional positional arguments to pass to the initializer. - return cls.from_trytes( - # :see: http://stackoverflow.com/a/1751478/ - (trits[i:i+3] for i in range(0, len(trits), 3)), + :param kwargs: + Additional keyword arguments to pass to the initializer. - *args, - **kwargs - ) + References: - def __init__(self, trytes, pad=None): - # type: (TrytesCompatible, Optional[int]) -> None - """ - :param trytes: - Byte string or bytearray. + - :py:func:`int_from_trits` + - :py:meth:`as_trits` + """ + # Allow passing a generator or other non-Sized value to this + # method. + trits = list(trits) - :param pad: - Ensure at least this many trytes. + if len(trits) % 3: + # Pad the trits so that it is cleanly divisible into trytes. + trits += [0] * (3 - (len(trits) % 3)) - If there are too few, null trytes will be appended to the - TryteString. + return cls.from_trytes( + # http://stackoverflow.com/a/1751478/ + (trits[i:i + 3] for i in range(0, len(trits), 3)), - Note: If the TryteString is too long, it will _not_ be - truncated! - """ - super(TryteString, self).__init__() - - if isinstance(trytes, (int, float)): - raise with_context( - exc = TypeError( - 'Converting {type} is not supported; ' - '{cls} is not a numeric type.'.format( - type = type(trytes).__name__, - cls = type(self).__name__, - ), - ), - - context = { - 'trytes': trytes, - }, - ) - - if isinstance(trytes, TryteString): - incoming_type = type(trytes) - - if incoming_type is TryteString or issubclass(incoming_type, type(self)): - # Create a copy of the incoming TryteString's trytes, to ensure - # we don't modify it when we apply padding. - trytes = bytearray(trytes._trytes) - - else: - raise with_context( - exc = TypeError( - '{cls} cannot be initialized from a(n) {type}.'.format( - type = type(trytes).__name__, - cls = type(self).__name__, - ), - ), + *args, + **kwargs + ) - context = { - 'trytes': trytes, - }, + def __init__(self, trytes, pad=None): + # type: (TrytesCompatible, Optional[int]) -> None + """ + :param trytes: + Byte string or bytearray. + + :param pad: + Ensure at least this many trytes. + + If there are too few, null trytes will be appended to the + TryteString. + + .. note:: + If the TryteString is too long, it will *not* be + truncated! + """ + super(TryteString, self).__init__() + + if isinstance(trytes, (int, float)): + raise with_context( + exc=TypeError( + 'Converting {type} is not supported; ' + '{cls} is not a numeric type.'.format( + type=type(trytes).__name__, + cls=type(self).__name__, + ), + ), + + context={ + 'trytes': trytes, + }, + ) + + if isinstance(trytes, TryteString): + incoming_type = type(trytes) + + if ( + (incoming_type is TryteString) or + issubclass(incoming_type, type(self)) + ): + # Create a copy of the incoming TryteString's trytes, to + # ensure we don't modify it when we apply padding. + trytes = bytearray(trytes._trytes) + + else: + raise with_context( + exc=TypeError( + '{cls} cannot be initialized from a(n) {type}.'.format( + type=type(trytes).__name__, + cls=type(self).__name__, + ), + ), + + context={ + 'trytes': trytes, + }, + ) + + else: + if isinstance(trytes, text_type): + trytes = encode(trytes, 'ascii') + + if not isinstance(trytes, bytearray): + trytes = bytearray(trytes) + + for i, ordinal in enumerate(trytes): + if ordinal not in AsciiTrytesCodec.index: + raise with_context( + exc=ValueError( + 'Invalid character {char!r} at position {i} ' + '(expected A-Z or 9).'.format( + char=chr(ordinal), + i=i, + ), + ), + + context={ + 'trytes': trytes, + }, + ) + + if pad: + trytes += b'9' * max(0, pad - len(trytes)) + + self._trytes = trytes # type: bytearray + + def __hash__(self): + # type: () -> int + return hash(binary_type(self._trytes)) + + def __repr__(self): + # type: () -> Text + return '{cls}({trytes!r})'.format( + cls=type(self).__name__, + trytes=binary_type(self._trytes), ) - else: - if isinstance(trytes, text_type): - trytes = encode(trytes, 'ascii') - - if not isinstance(trytes, bytearray): - trytes = bytearray(trytes) - - for i, ordinal in enumerate(trytes): - if ordinal not in AsciiTrytesCodec.index: - raise with_context( - exc = ValueError( - 'Invalid character {char!r} at position {i} ' - '(expected A-Z or 9).'.format( - char = chr(ordinal), - i = i, - ), - ), + def __bytes__(self): + """ + Converts the TryteString into an ASCII representation. + + .. note:: + This does not decode the trytes into bytes/characters; it + only returns an ASCII representation of the trytes + themselves! + + If you want to... + + - ... encode trytes into bytes: use :py:meth:`encode`. + - ... decode trytes into Unicode: use :py:meth:`decode`. + """ + return binary_type(self._trytes) + + def __str__(self): + """ + Same as :py:meth:`__bytes__`, except this method returns a + unicode string. + """ + # This causes infinite recursion in Python 2. + # return binary_type(self).decode('ascii') + + return binary_type(self._trytes).decode('ascii') + + def __bool__(self): + # type: () -> bool + return bool(self._trytes) and any(t != b'9' for t in self) + + if PY2: + # Magic methods have different names in Python 2. + __nonzero__ = __bool__ + + def __len__(self): + # type: () -> int + return len(self._trytes) + + def __iter__(self): + # type: () -> Generator[binary_type, None, None] + # :see: http://stackoverflow.com/a/14267935/ + return (binary_type(self._trytes[i:i + 1]) for i in range(len(self))) + + def __contains__(self, other): + # type: (TrytesCompatible) -> bool + if isinstance(other, TryteString): + return other._trytes in self._trytes + elif isinstance(other, text_type): + return other.encode('ascii') in self._trytes + elif isinstance(other, (binary_type, bytearray)): + return other in self._trytes + else: + raise with_context( + exc=TypeError( + 'Invalid type for TryteString contains check ' + '(expected Union[TryteString, {binary_type}, bytearray], ' + 'actual {type}).'.format( + binary_type=binary_type.__name__, + type=type(other).__name__, + ), + ), + + context={ + 'other': other, + }, + ) + + def __getitem__(self, item): + # type: (Union[int, slice]) -> TryteString + new_trytes = bytearray() + + sliced = self._trytes[item] + + if isinstance(sliced, int): + new_trytes.append(sliced) + else: + new_trytes.extend(sliced) + + return TryteString(new_trytes) + + def __setitem__(self, item, trytes): + # type: (Union[int, slice], TrytesCompatible) -> None + new_trytes = TryteString(trytes) + + if isinstance(item, slice): + self._trytes[item] = new_trytes._trytes + elif len(new_trytes) > 1: + raise with_context( + exc=ValueError( + 'Cannot assign multiple trytes to the same index ' + '(``exc.context`` has more info).' + ), + + context={ + 'self': self, + 'index': item, + 'new_trytes': new_trytes, + }, + ) + else: + self._trytes[item] = new_trytes._trytes[0] + + def __add__(self, other): + # type: (TrytesCompatible) -> TryteString + if isinstance(other, TryteString): + return TryteString(self._trytes + other._trytes) + elif isinstance(other, text_type): + return TryteString(self._trytes + other.encode('ascii')) + elif isinstance(other, (binary_type, bytearray)): + return TryteString(self._trytes + other) + else: + raise with_context( + exc=TypeError( + 'Invalid type for TryteString concatenation ' + '(expected Union[TryteString, {binary_type}, bytearray], ' + 'actual {type}).'.format( + binary_type=binary_type.__name__, + type=type(other).__name__, + ), + ), + + context={ + 'other': other, + }, + ) + + def __eq__(self, other): + # type: (TrytesCompatible) -> bool + if isinstance(other, TryteString): + return self._trytes == other._trytes + elif isinstance(other, text_type): + return self._trytes == other.encode('ascii') + elif isinstance(other, (binary_type, bytearray)): + return self._trytes == other + else: + raise with_context( + exc=TypeError( + 'Invalid type for TryteString comparison ' + '(expected Union[TryteString, {binary_type}, bytearray], ' + 'actual {type}).'.format( + binary_type=binary_type.__name__, + type=type(other).__name__, + ), + ), + + context={ + 'other': other, + }, + ) + + # :bc: In Python 2 this must be defined explicitly. + def __ne__(self, other): + # type: (TrytesCompatible) -> bool + return not (self == other) + + def count_chunks(self, chunk_size): + # type: (int) -> int + """ + Returns the number of constant-size chunks the TryteString can + be divided into (rounded up). + + :param chunk_size: + Number of trytes per chunk. + """ + return len(self.iter_chunks(chunk_size)) + + def iter_chunks(self, chunk_size): + # type: (int) -> ChunkIterator + """ + Iterates over the TryteString, in chunks of constant size. + + :param chunk_size: + Number of trytes per chunk. + + The final chunk will be padded if it is too short. + """ + return ChunkIterator(self, chunk_size) + + def encode(self, errors='strict', codec=AsciiTrytesCodec.name): + # type: (Text, Text) -> binary_type + """ + Encodes the TryteString into a lower-level primitive (usually + bytes). + + :param errors: + How to handle trytes that can't be converted: + + 'strict' + raise an exception (recommended). + + 'replace' + replace with '?'. + + 'ignore' + omit the tryte from the result. + + :param codec: + Reserved for future use. + + See https://github.com/iotaledger/iota.lib.py/issues/62 for + more information. + + :raise: + - :py:class:`iota.codecs.TrytesDecodeError` if the trytes + cannot be decoded into bytes. + """ + # Converting ASCII-encoded trytes into bytes is considered to be + # a *decode* operation according to + # :py:class:`AsciiTrytesCodec`. + # + # Once we add more codecs, we may need to revisit this. + # See https://github.com/iotaledger/iota.lib.py/issues/62 for + # more information. + return decode(self._trytes, codec, errors) + + def as_bytes(self, *args, **kwargs): + """ + Deprecated; use :py:meth:`encode` instead. + + https://github.com/iotaledger/iota.lib.py/issues/90 + """ + warn( + category=DeprecationWarning, + message='`as_bytes()` is deprecated; use `encode()` instead.', + ) + return self.encode(*args, **kwargs) - context = { - 'trytes': trytes, - }, - ) + def decode(self, errors='strict', strip_padding=True): + # type: (Text, bool) -> Text + """ + Decodes the TryteString into a higher-level abstraction (usually + Unicode characters). - if pad: - trytes += b'9' * max(0, pad - len(trytes)) + :param errors: + How to handle trytes that can't be converted, or bytes that can't + be decoded using UTF-8: - self._trytes = trytes # type: bytearray + 'strict' + raise an exception (recommended). - def __hash__(self): - # type: () -> int - return hash(binary_type(self._trytes)) + 'replace' + replace with a placeholder character. - def __repr__(self): - # type: () -> Text - return '{cls}({trytes!r})'.format( - cls = type(self).__name__, - trytes = binary_type(self._trytes), - ) + 'ignore' + omit the invalid tryte/byte sequence. - def __bytes__(self): - """ - Converts the TryteString into an ASCII representation. + :param strip_padding: + Whether to strip trailing null trytes before converting. - Note: This does not decode the trytes into bytes/characters; it - only returns an ASCII representation of the trytes themselves! + :raise: + - :py:class:`iota.codecs.TrytesDecodeError` if the trytes + cannot be decoded into bytes. + - :py:class:`UnicodeDecodeError` if the resulting bytes + cannot be decoded using UTF-8. + """ + trytes = self._trytes + if strip_padding and (trytes[-1] == ord(b'9')): + trytes = trytes.rstrip(b'9') - If you want to... - - ... encode trytes into bytes: use :py:meth:`encode`. - - ... decode trytes into Unicode: use :py:meth:`decode`. - """ - return binary_type(self._trytes) + # Put one back to preserve even length for ASCII codec. + trytes += b'9' * (len(trytes) % 2) - def __str__(self): - """ - Same as :py:meth:`__bytes__`, except this method returns a unicode - string. - """ - # This causes infinite recursion in Python 2. - # return binary_type(self).decode('ascii') - return binary_type(self._trytes).decode('ascii') - - def __bool__(self): - # type: () -> bool - return bool(self._trytes) and any(t != b'9' for t in self) - - if PY2: - # Magic methods have different names in Python 2. - __nonzero__ = __bool__ - - def __len__(self): - # type: () -> int - return len(self._trytes) - - def __iter__(self): - # type: () -> Generator[binary_type, None, None] - # :see: http://stackoverflow.com/a/14267935/ - return (binary_type(self._trytes[i:i + 1]) for i in range(len(self))) - - def __contains__(self, other): - # type: (TrytesCompatible) -> bool - if isinstance(other, TryteString): - return other._trytes in self._trytes - elif isinstance(other, text_type): - return other.encode('ascii') in self._trytes - elif isinstance(other, (binary_type, bytearray)): - return other in self._trytes - else: - raise with_context( - exc = TypeError( - 'Invalid type for TryteString contains check ' - '(expected Union[TryteString, {binary_type}, bytearray], ' - 'actual {type}).'.format( - binary_type = binary_type.__name__, - type = type(other).__name__, - ), - ), - - context = { - 'other': other, - }, - ) - - def __getitem__(self, item): - # type: (Union[int, slice]) -> TryteString - new_trytes = bytearray() - - sliced = self._trytes[item] - - if isinstance(sliced, int): - new_trytes.append(sliced) - else: - new_trytes.extend(sliced) - - return TryteString(new_trytes) - - def __setitem__(self, item, trytes): - # type: (Union[int, slice], TrytesCompatible) -> None - new_trytes = TryteString(trytes) - - if isinstance(item, slice): - self._trytes[item] = new_trytes._trytes - elif len(new_trytes) > 1: - raise with_context( - exc = ValueError( - 'Cannot assign multiple trytes to the same index ' - '(``exc.context`` has more info).' - ), - - context = { - 'self': self, - 'index': item, - 'new_trytes': new_trytes, - }, - ) - else: - self._trytes[item] = new_trytes._trytes[0] - - def __add__(self, other): - # type: (TrytesCompatible) -> TryteString - if isinstance(other, TryteString): - return TryteString(self._trytes + other._trytes) - elif isinstance(other, text_type): - return TryteString(self._trytes + other.encode('ascii')) - elif isinstance(other, (binary_type, bytearray)): - return TryteString(self._trytes + other) - else: - raise with_context( - exc = TypeError( - 'Invalid type for TryteString concatenation ' - '(expected Union[TryteString, {binary_type}, bytearray], ' - 'actual {type}).'.format( - binary_type = binary_type.__name__, - type = type(other).__name__, - ), - ), - - context = { - 'other': other, - }, - ) - - def __eq__(self, other): - # type: (TrytesCompatible) -> bool - if isinstance(other, TryteString): - return self._trytes == other._trytes - elif isinstance(other, text_type): - return self._trytes == other.encode('ascii') - elif isinstance(other, (binary_type, bytearray)): - return self._trytes == other - else: - raise with_context( - exc = TypeError( - 'Invalid type for TryteString comparison ' - '(expected Union[TryteString, {binary_type}, bytearray], ' - 'actual {type}).'.format( - binary_type = binary_type.__name__, - type = type(other).__name__, - ), - ), - - context = { - 'other': other, - }, - ) - - # :bc: In Python 2 this must be defined explicitly. - def __ne__(self, other): - # type: (TrytesCompatible) -> bool - return not (self == other) - - def count_chunks(self, chunk_size): - # type: (int) -> int - """ - Returns the number of constant-size chunks the TryteString can be - divided into (rounded up). + bytes_ = decode(trytes, AsciiTrytesCodec.name, errors) - :param chunk_size: - Number of trytes per chunk. - """ - return len(self.iter_chunks(chunk_size)) + return bytes_.decode('utf-8', errors) - def iter_chunks(self, chunk_size): - # type: (int) -> ChunkIterator - """ - Iterates over the TryteString, in chunks of constant size. + def as_string(self, *args, **kwargs): + """ + Deprecated; use :py:meth:`decode` instead. - :param chunk_size: - Number of trytes per chunk. - The final chunk will be padded if it is too short. - """ - return ChunkIterator(self, chunk_size) + https://github.com/iotaledger/iota.lib.py/issues/90 + """ + warn( + category=DeprecationWarning, + message='`as_string()` is deprecated; use `decode()` instead.', + ) + return self.decode(*args, **kwargs) - def encode(self, errors='strict', codec=AsciiTrytesCodec.name): - # type: (Text, Text) -> binary_type - """ - Encodes the TryteString into a lower-level primitive (usually - bytes). + def as_json_compatible(self): + # type: () -> Text + """ + Returns a JSON-compatible representation of the object. - :param errors: - How to handle trytes that can't be converted: - - 'strict': raise an exception (recommended). - - 'replace': replace with '?'. - - 'ignore': omit the tryte from the result. + References: - :param codec: - Reserved for future use. + - :py:class:`iota.json.JsonEncoder`. + """ + return self._trytes.decode('ascii') - See https://github.com/iotaledger/iota.lib.py/issues/62 for more - information. + def as_integers(self): + # type: () -> List[int] + """ + Converts the TryteString into a sequence of integers. - :raise: - - :py:class:`iota.codecs.TrytesDecodeError` if the trytes cannot - be decoded into bytes. - """ - # Converting ASCII-encoded trytes into bytes is considered to be a - # *decode* operation according to :py:class:`AsciiTrytesCodec`. - # Once we add more codecs, we may need to revisit this. - # See https://github.com/iotaledger/iota.lib.py/issues/62 for more - # information. - # - # In Python 2, :py:func:`decode` does not accept keyword arguments. - return decode(self._trytes, codec, errors) - - def as_bytes(self, *args, **kwargs): - """ - Deprecated; use :py:meth:`encode` instead. + Each integer is a value between -13 and 13. + """ + return [ + self._normalize(AsciiTrytesCodec.index[c]) + for c in self._trytes + ] - https://github.com/iotaledger/iota.lib.py/issues/90 - """ - warn( - category=DeprecationWarning, - message='`as_bytes()` is deprecated; use `encode()` instead.', - ) - return self.encode(*args, **kwargs) - - def decode(self, errors='strict', strip_padding=True): - # type: (Text, bool) -> Text - """ - Decodes the TryteString into a higher-level abstraction (usually - Unicode characters). - - :param errors: - How to handle trytes that can't be converted, or bytes that can't - be decoded using UTF-8: - - 'strict': raise an exception (recommended). - - 'replace': replace with a placeholder character. - - 'ignore': omit the invalid tryte/byte sequence. - - :param strip_padding: - Whether to strip trailing null trytes before converting. - - :raise: - - :py:class:`iota.codecs.TrytesDecodeError` if the trytes cannot - be decoded into bytes. - - :py:class:`UnicodeDecodeError` if the resulting bytes cannot be - decoded using UTF-8. - """ - trytes = self._trytes - if strip_padding and (trytes[-1] == ord(b'9')): - trytes = trytes.rstrip(b'9') + def as_trytes(self): + # type: () -> List[List[int]] + """ + Converts the TryteString into a sequence of trytes. - # Put one back to preserve even length for ASCII codec. - trytes += b'9' * (len(trytes) % 2) + Each tryte is represented as a list with 3 trit values. - return decode(trytes, AsciiTrytesCodec.name, errors).decode('utf-8', errors) + See :py:meth:`as_trits` for more info. - def as_string(self, *args, **kwargs): - """ - Deprecated; use :py:meth:`decode` instead. + .. important:: + :py:class:`TryteString` is not a numeric type, so the result + of this method should not be interpreted as an integer! + """ + return [ + trits_from_int(n, pad=3) + for n in self.as_integers() + ] - https://github.com/iotaledger/iota.lib.py/issues/90 - """ - warn( - category=DeprecationWarning, - message='`as_string()` is deprecated; use `decode()` instead.', - ) - return self.decode(*args, **kwargs) - - def as_json_compatible(self): - # type: () -> Text - """ - Returns a JSON-compatible representation of the object. + def as_trits(self): + # type: () -> List[int] + """ + Converts the TryteString into a sequence of trit values. - References: - - :py:class:`iota.json.JsonEncoder`. - """ - return self._trytes.decode('ascii') + A trit may have value 1, 0, or -1. - def as_integers(self): - # type: () -> List[int] - """ - Converts the TryteString into a sequence of integers. + References: - Each integer is a value between -13 and 13. - """ - return [ - self._normalize(AsciiTrytesCodec.index[c]) - for c in self._trytes - ] + - https://en.wikipedia.org/wiki/Balanced_ternary - def as_trytes(self): - # type: () -> List[List[int]] - """ - Converts the TryteString into a sequence of trytes. + .. important:: + :py:class:`TryteString` is not a numeric type, so the result + of this method should not be interpreted as an integer! + """ + # http://stackoverflow.com/a/952952/5568265#comment4204394_952952 + return list(chain.from_iterable(self.as_trytes())) - Each tryte is represented as a list with 3 trit values. + def _repr_pretty_(self, p, cycle): + """ + Makes JSON-serializable objects play nice with IPython's default + pretty-printer. - See :py:meth:`as_trits` for more info. + Sadly, :py:func:`pprint.pprint` does not have a similar + mechanism. - IMPORTANT: TryteString is not a numeric type, so the result of this - method should not be interpreted as an integer! - """ - return [ - trits_from_int(n, pad=3) - for n in self.as_integers() - ] + References: - def as_trits(self): - # type: () -> List[int] - """ - Converts the TryteString into a sequence of trit values. + - http://ipython.readthedocs.io/en/stable/api/generated/IPython.lib.pretty.html + - :py:meth:`IPython.lib.pretty.RepresentationPrinter.pretty` + - :py:func:`pprint._safe_repr` + """ + return p.text(repr(self)) - A trit may have value 1, 0, or -1. + @staticmethod + def _normalize(n): + # type: (int) -> int + if n > 26: + raise ValueError( + '{n} cannot be represented by a single tryte.'.format( + n=n, + )) - References: - - https://en.wikipedia.org/wiki/Balanced_ternary + # For values greater than 13, trigger an overflow. + # E.g., 14 => -13, 15 => -12, etc. + return (n - 27) if n > 13 else n - IMPORTANT: TryteString is not a numeric type, so the result of this - method should not be interpreted as an integer! - """ - # http://stackoverflow.com/a/952952/5568265#comment4204394_952952 - return list(chain.from_iterable(self.as_trytes())) - def _repr_pretty_(self, p, cycle): +class ChunkIterator(Iterator[TryteString]): """ - Makes JSON-serializable objects play nice with IPython's default - pretty-printer. - - Sadly, :py:func:`pprint.pprint` does not have a similar mechanism. - - References: - - http://ipython.readthedocs.io/en/stable/api/generated/IPython.lib.pretty.html - - :py:meth:`IPython.lib.pretty.RepresentationPrinter.pretty` - - :py:func:`pprint._safe_repr` + Iterates over a TryteString, in chunks of constant size. """ - return p.text(repr(self)) - @staticmethod - def _normalize(n): - # type: (int) -> int - if n > 26: - raise ValueError('{n} cannot be represented by a single tryte.'.format( - n = n, - )) + def __init__(self, trytes, chunk_size): + # type: (TryteString, int) -> None + """ + :param trytes: + :py:class:`TryteString` to iterate over. - # For values greater than 13, trigger an overflow. - # E.g., 14 => -13, 15 => -12, etc. - return (n - 27) if n > 13 else n + :param chunk_size: + Number of trytes per chunk. + The final chunk will be padded if it is too short. + """ + super(ChunkIterator, self).__init__() -class ChunkIterator(Iterator[TryteString]): - """ - Iterates over a TryteString, in chunks of constant size. - """ - def __init__(self, trytes, chunk_size): - # type: (TryteString, int) -> None - """ - :param trytes: - TryteString to iterate over. + self.trytes = trytes + self.chunk_size = chunk_size - :param chunk_size: - Number of trytes per chunk. - The final chunk will be padded if it is too short. - """ - super(ChunkIterator, self).__init__() - - self.trytes = trytes - self.chunk_size = chunk_size + self._offset = 0 - self._offset = 0 + def __iter__(self): + # type: () -> ChunkIterator + return self - def __iter__(self): - # type: () -> ChunkIterator - return self + def __len__(self): + # type: () -> int + """ + Returns how many chunks this iterator will return. - def __len__(self): - # type: () -> int - """ - Returns how many chunks this iterator will return. - - Note: This method always returns the same result, no matter how - many iterations have been completed. - """ - return int(ceil(len(self.trytes) / self.chunk_size)) + .. note:: + This method always returns the same result, no matter how + many iterations have been completed. + """ + return int(ceil(len(self.trytes) / self.chunk_size)) - def __next__(self): - # type: () -> TryteString - """ - Returns the next chunk in the iterator. + def __next__(self): + # type: () -> TryteString + """ + Returns the next chunk in the iterator. - :raise: - - :py:class:`StopIteration` if there are no more chunks - available. - """ - if self._offset >= len(self.trytes): - raise StopIteration + :raise: + - :py:class:`StopIteration` if there are no more chunks + available. + """ + if self._offset >= len(self.trytes): + raise StopIteration - chunk = self.trytes[self._offset:self._offset+self.chunk_size] - chunk += b'9' * max(0, self.chunk_size - len(chunk)) + chunk = self.trytes[self._offset:self._offset + self.chunk_size] + chunk += b'9' * max(0, self.chunk_size - len(chunk)) - self._offset += self.chunk_size + self._offset += self.chunk_size - return chunk + return chunk - if PY2: - # In Python 2, iterator methods are named a little differently. - next = __next__ + if PY2: + # In Python 2, iterator methods are named a little differently. + next = __next__ class Hash(TryteString): - """ - A TryteString that is exactly one hash long. - """ - # Divide by 3 to convert trits to trytes. - LEN = HASH_LENGTH // TRITS_PER_TRYTE + """ + A TryteString that is exactly one hash long. + """ + # Divide by 3 to convert trits to trytes. + LEN = HASH_LENGTH // TRITS_PER_TRYTE - def __init__(self, trytes): - # type: (TrytesCompatible) -> None - super(Hash, self).__init__(trytes, pad=self.LEN) + def __init__(self, trytes): + # type: (TrytesCompatible) -> None + super(Hash, self).__init__(trytes, pad=self.LEN) - if len(self._trytes) > self.LEN: - raise with_context( - exc = ValueError('{cls} values must be {len} trytes long.'.format( - cls = type(self).__name__, - len = self.LEN - )), + if len(self._trytes) > self.LEN: + raise with_context( + exc=ValueError('{cls} values must be {len} trytes long.'.format( + cls=type(self).__name__, + len=self.LEN + )), - context = { - 'trytes': trytes, - }, - ) + context={ + 'trytes': trytes, + }, + ) class Address(TryteString): - """ - A TryteString that acts as an address, with support for generating - and validating checksums. - """ - LEN = Hash.LEN - - def __init__(self, trytes, balance=None, key_index=None, security_level=None): - # type: (TrytesCompatible, Optional[int], Optional[int], Optional[int]) -> None - super(Address, self).__init__(trytes, pad=self.LEN) - - self.checksum = None - if len(self._trytes) == (self.LEN + AddressChecksum.LEN): - self.checksum = AddressChecksum(self[self.LEN:]) # type: Optional[AddressChecksum] - - elif len(self._trytes) > self.LEN: - raise with_context( - exc = ValueError( - 'Address values must be {len_no_checksum} trytes (no checksum), ' - 'or {len_with_checksum} trytes (with checksum).'.format( - len_no_checksum = self.LEN, - len_with_checksum = self.LEN + AddressChecksum.LEN, - ), - ), - - context = { - 'trytes': trytes, - }, - ) - - # Make the address sans checksum accessible. - self.address = self[:self.LEN] # type: TryteString - - self.balance = balance - """ - Balance owned by this address. - Must be set manually via the ``getInputs`` command. - - References: - - :py:class:`iota.commands.extended.get_inputs` - - :py:meth:`ProposedBundle.add_inputs` """ + A TryteString that acts as an address, with support for generating + and validating checksums. + """ + LEN = Hash.LEN + + def __init__( + self, + trytes, # type: TrytesCompatible + balance=None, # type: Optional[int] + key_index=None, # type: Optional[int] + security_level=None, # type: Optional[int] + ): + # type: (...) -> None + super(Address, self).__init__(trytes, pad=self.LEN) + + self.checksum = None + if len(self._trytes) == (self.LEN + AddressChecksum.LEN): + self.checksum = AddressChecksum( + self[self.LEN:] + ) # type: Optional[AddressChecksum] + + elif len(self._trytes) > self.LEN: + raise with_context( + exc=ValueError( + 'Address values must be ' + '{len_no_checksum} trytes (no checksum), ' + 'or {len_with_checksum} trytes (with checksum).'.format( + len_no_checksum=self.LEN, + len_with_checksum=self.LEN + AddressChecksum.LEN, + ), + ), + + context={ + 'trytes': trytes, + }, + ) + + # Make the address sans checksum accessible. + self.address = self[:self.LEN] # type: TryteString + + self.balance = balance + """ + Balance owned by this address. + Defaults to ``None``; usually set via the ``getInputs`` command. + + References: + + - :py:class:`iota.commands.extended.get_inputs` + - :py:meth:`ProposedBundle.add_inputs` + """ + + self.key_index = key_index + """ + Index of the key used to generate this address. + Defaults to ``None``; usually set via ``AddressGenerator``. + + References: + + - :py:class:`iota.crypto.addresses.AddressGenerator` + """ + + self.security_level = security_level + """ + Number of hashes in the digest that was used to generate this + address. + """ + + def as_json_compatible(self): + # type: () -> dict + return { + 'trytes': self._trytes.decode('ascii'), + 'balance': self.balance, + 'key_index': self.key_index, + 'security_level': self.security_level, + } + + def is_checksum_valid(self): + # type: () -> bool + """ + Returns whether this address has a valid checksum. + """ + if self.checksum: + return self.checksum == self._generate_checksum() + + return False + + def with_valid_checksum(self): + # type: () -> Address + """ + Returns the address with a valid checksum attached. + """ + return Address( + trytes=self.address + self._generate_checksum(), + + # Make sure to copy all of the ancillary attributes, too! + balance=self.balance, + key_index=self.key_index, + security_level=self.security_level, + ) - self.key_index = key_index - """ - Index of the key used to generate this address. - Must be set manually via ``AddressGenerator``. + def _generate_checksum(self): + # type: () -> AddressChecksum + """ + Generates the correct checksum for this address. + """ + checksum_trits = [] # type: MutableSequence[int] - References: - - :py:class:`iota.crypto.addresses.AddressGenerator` - """ + sponge = Kerl() + sponge.absorb(self.address.as_trits()) + sponge.squeeze(checksum_trits) - self.security_level = security_level - """ - Number of hashes in the digest that was used to generate this - address. - """ + checksum_length = AddressChecksum.LEN * TRITS_PER_TRYTE - def as_json_compatible(self): - # type: () -> dict - return { - 'trytes': self._trytes.decode('ascii'), - 'balance': self.balance, - 'key_index': self.key_index, - 'security_level': self.security_level, - } - - def is_checksum_valid(self): - # type: () -> bool - """ - Returns whether this address has a valid checksum. - """ - if self.checksum: - return self.checksum == self._generate_checksum() + return AddressChecksum.from_trits(checksum_trits[-checksum_length:]) - return False - def with_valid_checksum(self): - # type: () -> Address +class AddressChecksum(TryteString): """ - Returns the address with a valid checksum attached. + A TryteString that acts as an address checksum. """ - return Address( - trytes = self.address + self._generate_checksum(), + LEN = 9 - # Make sure to copy all of the ancillary attributes, too! - balance = self.balance, - key_index = self.key_index, - security_level = self.security_level, - ) - - def _generate_checksum(self): - # type: () -> AddressChecksum - """ - Generates the correct checksum for this address. - """ - checksum_trits = [] # type: MutableSequence[int] + def __init__(self, trytes): + # type: (TrytesCompatible) -> None + super(AddressChecksum, self).__init__(trytes, pad=None) - sponge = Kerl() - sponge.absorb(self.address.as_trits()) - sponge.squeeze(checksum_trits) + if len(self._trytes) != self.LEN: + raise with_context( + exc=ValueError( + '{cls} values must be exactly {len} trytes long.'.format( + cls=type(self).__name__, + len=self.LEN, + ), + ), - checksum_length = AddressChecksum.LEN * TRITS_PER_TRYTE + context={ + 'trytes': trytes, + }, + ) - return AddressChecksum.from_trits(checksum_trits[-checksum_length:]) +class Tag(TryteString): + """ + A TryteString that acts as a transaction tag. + """ + LEN = 27 -class AddressChecksum(TryteString): - """ - A TryteString that acts as an address checksum. - """ - LEN = 9 - - def __init__(self, trytes): - # type: (TrytesCompatible) -> None - super(AddressChecksum, self).__init__(trytes, pad=None) - - if len(self._trytes) != self.LEN: - raise with_context( - exc = ValueError( - '{cls} values must be exactly {len} trytes long.'.format( - cls = type(self).__name__, - len = self.LEN, - ), - ), - - context = { - 'trytes': trytes, - }, - ) + def __init__(self, trytes): + # type: (TrytesCompatible) -> None + super(Tag, self).__init__(trytes, pad=self.LEN) + if len(self._trytes) > self.LEN: + raise with_context( + exc=ValueError('{cls} values must be {len} trytes long.'.format( + cls=type(self).__name__, + len=self.LEN + )), -class Tag(TryteString): - """ - A TryteString that acts as a transaction tag. - """ - LEN = 27 - - def __init__(self, trytes): - # type: (TrytesCompatible) -> None - super(Tag, self).__init__(trytes, pad=self.LEN) - - if len(self._trytes) > self.LEN: - raise with_context( - exc = ValueError('{cls} values must be {len} trytes long.'.format( - cls = type(self).__name__, - len = self.LEN - )), - - context = { - 'trytes': trytes, - }, - ) + context={ + 'trytes': trytes, + }, + )