"""Stores readline-style history and current place in it"""

def __init__(self, entries=None, duplicates=True, hist_size=100):

if entries is None:

self.entries = ['']

else:

self.entries = list(entries)

# how many lines back in history is currently selected where 0 is the

# saved typed line, 1 the prev entered line

self.index = 0

# what was on the prompt before using history

self.saved_line = ''

self.duplicates = duplicates

self.hist_size = hist_size

def append(self, line):

self.append_to(self.entries, line)

def append_to(self, entries, line):

line = line.rstrip('\n')

if line:

if not self.duplicates:

# remove duplicates

try:

while True:

entries.remove(line)

except ValueError:

pass

entries.append(line)

def first(self):

"""Move back to the beginning of the history."""

if not self.is_at_end:

self.index = len(self.entries)

return self.entries[-self.index]

def back(self, start=True, search=False, target=None,

include_current=False):

"""Move one step back in the history."""

if target is None:

target = self.saved_line

if not self.is_at_end:

if search:

self.index += self.find_partial_match_backward(target,

include_current)

elif start:

self.index += self.find_match_backward(target, include_current)

else:

self.index += 1

return self.entry

@property

def entry(self):

"""The current entry, which may be the saved line"""

return self.entries[-self.index] if self.index else self.saved_line

@property

def entries_by_index(self):

return list(reversed(self.entries + [self.saved_line]))

def find_match_backward(self, search_term, include_current=False):

add = 0 if include_current else 1

start = self.index + add

for idx, val in enumerate(islice(self.entries_by_index, start, None)):

if val.startswith(search_term):

return idx + add

return 0

def find_partial_match_backward(self, search_term, include_current=False):

add = 0 if include_current else 1

start = self.index + add

for idx, val in enumerate(islice(self.entries_by_index, start, None)):

if search_term in val:

return idx + add

return 0

def forward(self, start=True, search=False, target=None,

include_current=False):

"""Move one step forward in the history."""

if target is None:

target = self.saved_line

if self.index > 1:

if search:

self.index -= self.find_partial_match_forward(target,

include_current)

elif start:

self.index -= self.find_match_forward(target, include_current)

else:

self.index -= 1

return self.entry

else:

self.index = 0

return self.saved_line

def find_match_forward(self, search_term, include_current=False):

add = 0 if include_current else 1

end = max(0, self.index - (1 - add))

for idx in range(end):

val = self.entries_by_index[end - 1 - idx]

if val.startswith(search_term):

return idx + (0 if include_current else 1)

return self.index

def find_partial_match_forward(self, search_term, include_current=False):

add = 0 if include_current else 1

end = max(0, self.index - (1 - add))

for idx in range(end):

val = self.entries_by_index[end - 1 - idx]

if search_term in val:

return idx + add

return self.index

def last(self):

"""Move forward to the end of the history."""

if not self.is_at_start:

self.index = 0

return self.entries[0]

@property

def is_at_end(self):

return self.index >= len(self.entries) or self.index == -1

@property

def is_at_start(self):

return self.index == 0

def enter(self, line):

if self.index == 0:

self.saved_line = line

@classmethod

def from_filename(cls, filename):

history = cls()

history.load(filename)

return history

def reset(self):

self.index = 0

self.saved_line = ''

def load(self, filename, encoding):

with io.open(filename, 'r', encoding=encoding,

errors='ignore') as hfile:

with FileLock(hfile):

self.entries = self.load_from(hfile)

def load_from(self, fd):

entries = []

for line in fd:

self.append_to(entries, line)

return entries

def save(self, filename, encoding, lines=0):

fd = os.open(filename, os.O_WRONLY | os.O_CREAT | os.TRUNC,

stat.S_IRUSR | stat.S_IWUSR)

with io.open(fd, 'w', encoding=encoding,

errors='ignore') as hfile:

with FileLock(hfile):

self.save_to(hfile, self.entries, lines)

def save_to(self, fd, entries=None, lines=0):

if entries is None:

entries = self.entries

for line in entries[-lines:]:

fd.write(line)

fd.write('\n')

def append_reload_and_write(self, s, filename, encoding):

if not self.hist_size:

return self.append(s)

try:

fd = os.open(filename, os.O_APPEND | os.O_RDWR | os.O_CREAT,

stat.S_IRUSR | stat.S_IWUSR)

with io.open(fd, 'a+', encoding=encoding,

errors='ignore') as hfile:

with FileLock(hfile):

# read entries

hfile.seek(0, os.SEEK_SET)

entries = self.load_from(hfile)

self.append_to(entries, s)

# write new entries

hfile.seek(0, os.SEEK_SET)

hfile.truncate()

self.save_to(hfile, entries, self.hist_size)

self.entries = entries

except EnvironmentError as err:

raise RuntimeError(

_('Error occurred while writing to file %s (%s)')

% (filename, err.strerror))

else:

if len(self.entries) == 0:

# Make sure that entries contains at least one element. If the

# file and s are empty, this can occur.