#!/usr/bin/python """Cat a file at some speed. Optionally cats repeated text at a higher speed. Works at up to 50000 bps or so on my netbook. Above that it starts falling behind. """ import time def main(argv, stdin, stdout, stderr): if len(argv) in [2, 3]: baud = int(argv[1]) else: return usage(argv[0], stderr) if len(argv) == 3: fastbaud = int(argv[2]) else: fastbaud = baud cat(stdin, stdout, seconds_from_baud(baud), seconds_from_baud(fastbaud)) return 0 def usage(argv0, stderr): stderr.write("Usage: %s baud [fastbaud] < infile\n" % argv0) return 1 def seconds_from_baud(baud): return 1/(baud/10.0) assert seconds_from_baud(20) == 0.5 def cat(infile, outfile, seconds_per_char, seconds_per_repeated_char): timer = Timer(min_delay_per_char=seconds_per_repeated_char) history = HistoricalRecord() while True: chars = timer.how_many_chars() data = infile.read(chars) if not data: break outfile.write(data) outfile.flush() for char in data: if history.is_repeated(char, prefix_length=7): timer.should_delay(seconds_per_repeated_char) else: timer.should_delay(seconds_per_char) history.put(char) timer.sleep_appropriately() class Timer: def __init__(self, min_delay_per_char): self.goal = time.time() self.min_delay = min_delay_per_char self.chars_since_last_sleep = 0 def should_delay(self, seconds): self.goal += seconds self.chars_since_last_sleep += 1 def sleep_appropriately(self): duration = self.goal - time.time() if duration> 0: time.sleep(duration) self.chars_since_last_sleep = 0 def how_many_chars(self): return max(1, self.chars_since_last_sleep) class HistoricalRecord: """Maintains a historical record of a stream so as to distinguish repeated from nonrepeated text. """ def __init__(self): self.start_positions = {} self.last_three_chars = (None, None, None) self.history = [] def put(self, char): a, b, c = self.last_three_chars self.last_three_chars = b, c, char if self.last_three_chars not in self.start_positions: self.start_positions[self.last_three_chars] = [] self.history.append(char) self.start_positions[self.last_three_chars].append(len(self.history) - 3) def is_repeated(self, char, prefix_length): assert prefix_length>= 3 prefix = self.history[-prefix_length:] triple = tuple(prefix[:3]) locations = self.start_positions.get(triple) if locations: wholestring = ''.join(prefix) + char for location in locations: if self.is_repeated_at(wholestring, location): return True return False def is_repeated_at(self, string, location): return ''.join(self.history[location:location+len(string)]) == string if __name__ == '__main__': import sys sys.exit(main(sys.argv, sys.stdin, sys.stdout, sys.stderr))