6
\$\begingroup\$

What follows in an attempt at implementing Mark V. Shaney using contemporary Python. One question has already been asked while working on a generator in the code, but included here is the entire module. Are there any places that could be more idiomatic or Pythonic?

import collections
import functools
import itertools
import random


def pairwise(iterable, n=2):
    """Using a window of width n, iterate over items sourced from iterable."""
    iterators = itertools.tee(iterable, n)
    for move, iterator in enumerate(iterators):
        for _ in range(move):
            next(iterator, None)
    return zip(*iterators)


class RandomCounter:
    """RandomCounter(counter) -> RandomCounter instance"""

    def __init__(self, counter, choices=random.choices):
        """Initialize the instance with population and weight data."""
        population, weights = [], []
        for key, value in counter.items():
            population.append(key)
            weights.append(value)
        self.__population = tuple(population)
        self.__cum_weights = tuple(itertools.accumulate(weights))
        self.__choices = choices

    def __iter__(self):
        """Return the iterator object itself."""
        return self

    def __next__(self):
        """Return another completely random item from the counter."""
        return self.__choices(
            self.__population, cum_weights=self.__cum_weights
        )[0]


class MarkovChain:
    """MarkovChain(iterable, n) -> MarkovChain instance"""

    def __init__(self, iterable, n):
        """Initialize the instance by building a database of usable links."""
        links = {}
        for *root, suffix in pairwise(iterable, n):
            links.setdefault(tuple(root), collections.Counter())[suffix] += 1
        self.__links = {
            key: RandomCounter(value) for key, value in links.items()
        }

    def build_chain(self, start_point):
        """Iterate over items from the chain until a dead end is found."""
        if start_point not in self.__links:
            raise KeyError(f'could not find {start_point!r} in the links')
        yield from start_point
        while True:
            try:
                random_counter = self.__links[start_point]
            except KeyError:
                break
            else:
                item = next(random_counter)
                yield item
                prefix, *root = start_point
                root.append(item)
                start_point = tuple(root)


class SpecialDeque(collections.deque):
    """SpecialDeque([iterable[, maxlen]]) -> SpecialDeque instance"""

    @property
    def prefix(self):
        """Property allowing capture of all but last item in deque."""
        item = self.pop()
        value = tuple(self)
        self.append(item)
        return value

    @property
    def suffix(self):
        """Property allowing capture of all but first item in deque."""
        item = self.popleft()
        value = tuple(self)
        self.appendleft(item)
        return value


class MarkVShaney(MarkovChain):
    """MarkVShaney(iterable, n) -> MarkVShaney instance"""

    TERMINATORS = frozenset('!.;?')
    BAD_END = frozenset(';')
    NEW_END = functools.partial(random.choice, tuple(TERMINATORS - BAD_END))

    def __init__(self, iterable, n):
        """Initialize a MarkovChain while identifying proper start words."""
        if n < 2:
            raise ValueError('chain links may not be shorter than two')
        start_words = collections.Counter()
        super().__init__(self.__get_start_words(iterable, n, start_words), n)
        self.__start_words = RandomCounter(start_words)

    @classmethod
    def __get_start_words(cls, iterable, n, start_words):
        """Transparently yield from iterable while collecting start words."""
        buffer = SpecialDeque(maxlen=n)
        for count, item in enumerate(iterable, 1):
            yield item
            buffer.append(item)
            if count == n:
                start_words[buffer.prefix] += 1
            if count >= n and buffer[0][-1] in cls.TERMINATORS:
                start_words[buffer.suffix] += 1
        if len(buffer) < n:
            raise ValueError('iterable was too short to satisfy n')

    def build_chain(self, start_point=None):
        """Build a chain and select a proper start point if not provided."""
        if start_point is None:
            start_point = next(self.__start_words)
        yield from super().build_chain(start_point)

    def build_paragraph(self, clauses=1, good_start=False, good_end=False):
        """Generate some clauses that have a relationship with each other."""
        while True:
            iterator, paragraph, sentence = self.build_chain(), [], []
            while len(paragraph) < clauses:
                try:
                    word = next(iterator)
                except StopIteration:
                    break
                else:
                    sentence.append(word)
                    if word[-1] in self.TERMINATORS:
                        paragraph.append(' '.join(sentence))
                        sentence.clear()
            else:
                if good_start:
                    sentence = paragraph[0]
                    character = sentence[0]
                    if character.islower():
                        paragraph[0] = character.upper() + sentence[1:]
                if good_end:
                    sentence = paragraph[-1]
                    character = sentence[-1]
                    if character in self.BAD_END:
                        paragraph[-1] = sentence[:-1] + self.NEW_END()
                return paragraph

This sample program can be used to demonstrate the mess produced by the module. It only takes a little code to provide an approximation of words in a file that can be fed into the main class. Making a paragraph of nonsense and displaying it is similarly easy to accomplish.

import textwrap

import mvs


def main():
    with open('pg17625.txt') as file:
        source = file.read()
    generator = mvs.MarkVShaney(source.split(), 3)
    paragraph = generator.build_paragraph(3)
    print(textwrap.fill(' '.join(paragraph), 79))


if __name__ == '__main__':
    main()

The text shown below is originally sourced from Artificial Light which is located on Project Gutenberg. It can take several tries to produce a paragraph of sufficient quality to read. Its ability to understand punctuation and sentence structure is extremely limited in scope.

The reaction which takes place when water and to the burner under pressure is governed, in order to curb the cost of gas-lighting an exhibition of "Philosophical Fireworks" produced by manganese, nickel, selenium, and some of the closely associated rays are now made so that the mixture to light. In developing the enormous beam intensity would not be allowed to impinge upon a commercial scale. One report which bears the earmarks of authenticity is from ten minutes to an increase of 15 per cent.

\$\endgroup\$
  • 1
    \$\begingroup\$ Can you share an example of what is produced? \$\endgroup\$ – Simon Forsberg Nov 3 '17 at 19:21
  • \$\begingroup\$ @SimonForsberg Yes, an example program has been added along with its example output. \$\endgroup\$ – Noctis Skytower Nov 3 '17 at 20:27
2
\$\begingroup\$
  1. The name pairwise is fine when n=2 but for the general case I would prefer a name like windowise.

  2. Two initial underscores are intended to make names that don't collide when classes are combined using inheritance. They don't make names private, but instead make names unique by inserting the name of the class:

    >>> dir(RandomCounter({}))
    ['_RandomCounter__choices', '_RandomCounter__cum_weights', '_RandomCounter__population', ...]
    

    Unless you really do intend to avoid name clashes due to inheritance, there's no need to use the double underscore. Instead, it is conventional to use a single initial underscore for names that are not supposed to be used by callers.

  3. The class name RandomCounter (and the associated docstring) could be improved. This class isn't a counter, it's an iterator that yields an infinite random sequence of weighted choices.

  4. It's not clear why you allow the caller to pass in a function for the choice parameter to RandomCounter.__init__. This parameter is not actually used, and it is hard to see how anything other than random.choices could be used, so I would suggest removing it for simplicity.

  5. There is no need to convert the population and cumulative weights to tuples. The random.choices function accepts any sequence, not just tuples.

  6. Instead of constructing the population and weights by iterating over counter.items(), you could take advantage of the guarantee that "If keys, values and items views are iterated over with no intervening modifications to the dictionary, the order of items will directly correspond" and write:

    self._population = list(counter.keys())
    self._cum_weights = list(itertools.accumulate(counter.values()))
    
  7. When you find yourself writing a class with __iter__ and __next__ methods but nothing else, it's usually simpler to write a generator function. Not everything needs to be a class! In this case I would write:

    def weighted_random_iterator(counter):
        """Generate an infinite series of keys from the dictionary counter,
        chosen at random with weights given by the corresponding values.
    
        """
        population = list(counter.keys())
        cum_weights = list(itertools.accumulate(counter.values()))
        while True:
            yield random.choices(population, cum_weights=cum_weights)[0]
    
  8. In MarkovChain the awkward line:

    links.setdefault(tuple(root), collections.Counter())[suffix] += 1
    

    can be improved by using collections.defaultdict. First, set

    links = collections.defaultdict(collections.Counter)
    

    and then you can update it like this:

    links[tuple(root)][suffix] += 1
    
  9. I would prefer to avoid the special case at the beginning of build_chain. If the starting point can't be found in the links, then the chain will immediately hit a dead end, and it would seem reasonable to handle that the same as any other dead end.

  10. I would implement build_chain like this:

    def build_chain(self, root):
        """Generate items starting at root, until a dead end is reached."""
        yield from root
        while root in self._links:
            item = next(self._links[root])
            yield item
            root = root[1:] + (item,)
    

    Note the use of the + operator to build a tuple without having to convert to a list and back again.

  11. It seems to me that there is a problem in _get_start_words, namely that the start words might include terminators — this could happen if another terminator appeared fewer than \$n\$ words after the previous one. It might make more sense to throw away the accumulated words after a terminator and start over again, like this:

    @classmethod
    def _get_start_words(cls, iterable, n, start_words):
        """Transparently yield from iterable while collecting start words."""
        buffer = []
        for item in iterable:
            yield item
            if item[-1] in cls.TERMINATORS:
                buffer.clear()
            elif len(buffer) < n - 1:
                buffer.append(item)
                if len(buffer) == n - 1:
                    start_words[tuple(buffer)] += 1
    

    Note that in this version of the function we don't need SpecialDeque, or even a plain collections.deque: an ordinary list will do.

  12. For NEW_END, there's no need to use functools.partial, you can use lambda (and compute TERMINATORS - BAD_END just once, rather than once per call to NEW_END):

    GOOD_END = tuple(TERMINATORS - BAD_END)
    NEW_END = lambda: random.choice(GOOD_END)
    

    but since NEW_END is only used once, you might as well inline it at its single point of use:

    paragraph[-1] = sentence[:-1] + random.choice(self.GOOD_END)
    
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.