""" LineOrientedStateMachine.py State machine for processing text files on a line-by-line basis using regular expressions to determine transition to next state. Subclass LineOrientedStateMachine, determine states, then set up dictionaries for navigation using regular expressions as keys to states as values. Extract information from regular expression match objects, and accumulate processed information in 'cargo' object. Unmatched line accumulate in the cargo.buffer until the next state is reached. See ExampleStateMachine below. Developed from sample code in "Text Processing in Python" by David Mertz,pp. 274-280, which can be found at http://gnosis.cx/TPiP/chap4.txt. Code below processes Mertz's example identically. Jack Trainor 2008 """ import sys import string import re import StringIO class InitializationError(Exception): pass class StateMachine: """ from David Mertz """ def __init__(self): self.handlers = [] self.startState = None self.endStates = [] def addState(self, handler, endState=False): self.handlers.append(handler) if endState: self.endStates.append(handler) def setStart(self, handler): self.startState = handler def run(self, cargo=None): if not self.startState: raise InitializationError, "must call .setStart() before .run()" if not self.endStates: raise InitializationError, "at least one state must be an endState" handler = self.startState while True: (newState, cargo) = handler(cargo) if newState in self.endStates: newState(cargo) break elif newState not in self.handlers: raise RuntimeError, "Invalid target %s" % newState else: handler = newState return self class Cargo(object): def __init__(self): self.path = "" self.fp = None self.navigation = {} self.matchObject = None self.buffer = None self.errorMsg = "" class LineOrientedStateMachine(StateMachine): def __init__(self, fp): StateMachine.__init__(self) self.fp = fp def match(self, cargo, compiledRe, line): cargo.matchObject = compiledRe.match(line) return (cargo.matchObject != None) def readLoop(self, cargo): """ Keep reading lines until a change to next state is triggered """ cargo.buffer = [] next = () while not next: line = cargo.fp.readline() if not line: next = self.END, (cargo) else: for matchRe in cargo.navigation.keys(): if self.match(cargo, matchRe, line): state = cargo.navigation[matchRe] next = state, (cargo) break if not next: cargo.buffer.append(line) #print "BUFFER:", line return next def getErrorNext(self, cargo, errorMsg): cargo.errorMsg = errorMsg return (self.ERROR, (cargo)) def END(self, cargo): cargo.fp.close() print "Normal termination" def ERROR(self, cargo): cargo.fp.close() print "Error:", cargo.errorMsg class ExampleCargo(Cargo): def __init__(self): Cargo.__init__(self) self.company = "" self.amount = 0 class ExampleStateMachine(LineOrientedStateMachine): COMPANY_RE = re.compile(r"^>> (.*)\s*$") ORDER_RE = re.compile(r"^\s*(\w+)\s+(\d+)([Kk]*).*$") COMMENT_START_RE = re.compile(r"^\*(.*)$") COMMENT_END_RE = re.compile(r"^(.*)\*$") def __init__(self, fp): LineOrientedStateMachine.__init__(self, fp) self.addState(self.START) self.addState(self.COMPANY) self.addState(self.ORDER) self.addState(self.COMMENT_START) self.addState(self.COMMENT_END) self.addState(self.END, endState=True) self.addState(self.ERROR, endState=True) self.setStart(self.START) def START(self, cargo): cargo = ExampleCargo() cargo.fp = self.fp cargo.navigation = {self.COMPANY_RE : self.COMPANY, self.COMMENT_START_RE : self.COMMENT_START } next = self.readLoop(cargo) return next def COMPANY(self, cargo): cargo.company = cargo.matchObject.group(1) cargo.amount = 0 cargo.navigation = {self.COMPANY_RE : self.COMPANY, self.ORDER_RE : self.ORDER, self.COMMENT_START_RE : self.COMMENT_START } #print "COMPANY:", cargo.company next = self.readLoop(cargo) return next def ORDER(self, cargo): product = cargo.matchObject.group(1) quantity = getQuantity(cargo.matchObject.group(2) + cargo.matchObject.group(3)) price = getProductPrice(cargo.company, product) cargo.amount += quantity * price cargo.navigation = {self.COMPANY_RE : self.COMPANY, self.ORDER_RE : self.ORDER, self.COMMENT_START_RE : self.COMMENT_START } #print "ORDER:", product, quantity, price, cargo.amount next = self.readLoop(cargo) if (next[0] != self.ORDER): printInvoice(cargo.company, cargo.amount) return next def COMMENT_START(self, cargo): comment = cargo.matchObject.group(1) #print "COMMENT:", comment if comment and comment[-1] == "*": next = self.COMMENT_END, (cargo) else: cargo.navigation = {self.COMMENT_END_RE : self.COMMENT_END } next = self.readLoop(cargo) return next def COMMENT_END(self, cargo): commentLine = cargo.matchObject.group(1) cargo.navigation = {self.COMPANY_RE : self.COMPANY, self.ORDER_RE : self.ORDER, self.COMMENT_START_RE : self.COMMENT_START } next = self.readLoop(cargo) return next REPORT = """ MONTHLY REPORT -- April 2002 =================================================================== Rules: - Each buyer has price schedule for each item (func of quantity). - Each buyer has a discount schedule based on dollar totals. - Discounts are per-order (i.e. contiguous block) - Buyer listing starts with line containing ">>", then buyer name. - Item quantities have name-whitespace-number, one per line. - Comment sections begin with line starting with an asterisk, and ends with first line that ends with an asterisk. >> Acme Purchasing widgets 100 whatzits 1000 doodads 5000 dingdongs 20 * Note to Donald: The best contact for Acme is Debbie Franlin, at * 413-555-0001. Fallback is Sue Fong (call switchboard). * >> Megamart doodads 10k whatzits 5k >> Fly-by-Night Sellers widgets 500 whatzits 4 flazs 1000 * Note to Harry: Have Sales contact FbN for negotiations * * Known buyers: >> Acme >> Megamart >> Standard (default discounts) * *** LATE ADDITIONS *** >> Acme Purchasing widgets 500 (rush shipment)** """ #-- General support functions # Discount consists of dollar requirement and a percentage reduction # Each buyer can have an ascending series of discounts, the highest # one applicable to a month is used. discount_schedules = { "STANDARD" : [(5000,10),(10000,20),(15000,30),(20000,40)], "ACME" : [(1000,10),(5000,15),(10000,30),(20000,40)], "MEGAMART" : [(2000,10),(5000,20),(10000,25),(30000,50)], "BAGOBOLTS" : [(2500,10),(5000,15),(10000,25),(30000,50)], } item_prices = { "STANDARD" : {'widgets':1.0, 'whatzits':0.9, 'doodads':1.1, 'dingdongs':1.3, 'flazs':0.7}, "ACME" : {'widgets':0.9, 'whatzits':0.9, 'doodads':1.0, 'dingdongs':0.9, 'flazs':0.6}, "MEGAMART" : {'widgets':1.0, 'whatzits':0.8, 'doodads':1.0, 'dingdongs':1.2, 'flazs':0.7}, "BAGOBOLTS" : {'widgets':0.8, 'whatzits':0.9, 'doodads':1.1, 'dingdongs':1.3, 'flazs':0.5}, } discount_types = { "STANDARD" : "standard", "ACME" : "negotiated", "MEGAMART" : "negotiated" } def getCompanyId(company): if company.upper().find("ACME") >= 0: return "ACME" elif company.upper().find("MEGAMART") >= 0: return "MEGAMART" else: return "STANDARD" def getDiscountSchedule(company): return discount_schedules.get(getCompanyId(company), discount_schedules["STANDARD"]) def getProductPrice(company, product): prices = item_prices.get(getCompanyId(company), item_prices["STANDARD"]) return prices[product] def getDiscountType(company): return discount_types.get(getCompanyId(company), discount_types["STANDARD"]) def getDiscountedAmount(company, amount): multiplier = 1.0 for threshhold, percent in getDiscountSchedule(company): if amount >= threshhold: multiplier = 1 - float(percent)/100 return amount * multiplier def getQuantity(quantity): quantity = string.replace(string.upper(quantity),'K','000') quantity = int(quantity) return quantity def printInvoice(company, amount): print "Company name:", company, "(%s discounts)" % getDiscountType(company) print "Invoice total: $", getDiscountedAmount(company, amount), '\n' def main(): parser = ExampleStateMachine(StringIO.StringIO(REPORT)).run() if __name__ == "__main__": main()