Long overdue checkin of the state of code at the end of chapter 5

UltraBob · UltraBob · commit 37c932979619 · 2021-06-07T16:36:46.000+09:00
diff --git a/Work/fileparse.py b/Work/fileparse.py
@@ -1,3 +1,52 @@
 # fileparse.py
 #
 # Exercise 3.3
+import csv
+
+
+def parse_csv(lines, select=[], types=[], has_headers = True, delimiter=',', silence_errors=False) -> list:
+    """
+    Parse a CSV file into a list of records.
+    """
+    if select and not has_headers:
+        raise RuntimeError("Select argument requires column headers")
+    if isinstance(lines, str):
+        raise SystemExit("Expecting an iterable argument that can be parsed as csv.")
+
+    rows = csv.reader(lines, delimiter=delimiter)
+
+    # Read the file headers
+    headers = next(rows) if has_headers else []
+
+    # If a column selector was given, find indices of the specified columns.
+    # Also narrow the set of headers user for resulting dictionaries.
+    if select:
+        indices = [headers.index(colname) for colname in select]
+        headers = select
+
+    records = []
+    for row_no, row in enumerate(rows, start=1):
+        if not row:
+            continue
+        # Filter the row if specified columns were selected
+        if select:
+            row = [row[index] for index in indices]
+
+        if types:
+            try:
+                row = [func(val) for func, val in zip(types, row)]
+            except ValueError as e:
+                if silence_errors:
+                    continue
+                print(f'Row {row_no} Invalid: `{row}` cannot be converted.')
+                print(f'  Reason: {e}')
+
+        if headers:
+            # Make a dictionary
+            record = dict(zip(headers, row))
+        else:
+            # Make a tuple
+            record = tuple(row)
+        records.append(record)
+
+    return records
diff --git a/Work/pcost.py b/Work/pcost.py
@@ -1,11 +1,27 @@
+#!/usr/bin/env python3
 # pcost.py
 #
 # Exercise 1.27
-cost = 0
-with open('Data/portfolio.csv') as f:
-    next(f)
-    for line in f:
-        parts = line.split(',')
-        cost += int(parts[1]) * float(parts[2])
-
-print(f'Total cost {cost:.2f}')
+import os, csv, sys
+from report import read_portfolio
+
+
+def portfolio_cost(filename):
+    portfolio = read_portfolio(filename)
+    cost = 0
+    for holding in portfolio:
+        cost += holding.cost
+    return cost
+
+
+def main(argv):
+    if len(argv) != 2:
+        raise SystemExit(f'Usage: {argv[0]} portfoliofile')
+    filename = argv[1]
+    cost = portfolio_cost(filename)
+    print(f'Total cost {cost:.2f}')
+
+
+if __name__ == '__main__':
+    import sys
+    main(sys.argv)
diff --git a/Work/report.py b/Work/report.py
@@ -1,3 +1,72 @@
+#!/usr/bin/env python3
 # report.py
 #
 # Exercise 2.4
+import csv
+from fileparse import parse_csv
+import stock
+import tableformat
+
+
+def read_portfolio(filename: str) -> list:
+    """
+    Computes the total cost (shares*price) of a portfolio file
+    """
+    with open(filename, 'rt') as f:
+        holdings = parse_csv(f, types=[str, int, float])
+        portfolio = [stock.Stock(holding['name'], holding['shares'], holding['price']) for holding in holdings]
+    return portfolio
+
+
+def read_prices(filename: str) -> dict:
+    """
+    Read prices from a CSV file of name, price data
+    :param filename:
+    :return:
+    """
+    with open(filename, 'rt') as f:
+        prices = parse_csv(f, types=[str, float], has_headers=False)
+    return dict(prices)
+
+
+def make_report(portfolio, prices):
+    data = []
+    for holding in portfolio:
+        price = prices[holding.name]
+        data.append((holding.name, holding.shares, price, price - holding.price))
+    return data
+
+
+def print_report(report_data, formatter):
+    formatter.headings(['Name', 'Shares', 'Price', 'Change'])
+    for name, shares, price, change in report_data:
+        row_data = [name, str(shares), f'{price:0.2f}', f'{change:0.2f}']
+        formatter.row(row_data)
+
+
+def portfolio_report(portfolio_file, prices_file, fmt='txt'):
+    """
+    Make a stock report given portfolio and price data files.
+    """
+    # Read data files
+    portfolio = read_portfolio(portfolio_file)
+    prices = read_prices(prices_file)
+
+    # Create the report data
+    report = make_report(portfolio, prices)
+
+    # Print it out
+    formatter = tableformat.create_formatter(fmt)
+    print_report(report, formatter)
+
+
+def main(argv):
+    if len(argv) not in [3, 4]:
+        raise SystemExit(f'Usage: {argv[0]} portfoliofile pricefile')
+    print(argv)
+    portfolio_report(*argv[1:])
+
+
+if __name__ == '__main__':
+    import sys
+    main(sys.argv)
diff --git a/Work/stock.py b/Work/stock.py
@@ -0,0 +1,40 @@
+class Stock:
+    __slots__ = ('name', '_shares', 'price')
+    def __init__(self, name, shares, price):
+        self.name = name
+        self.shares = shares
+        self.price = price
+
+    @property
+    def cost(self):
+        return self.shares * self.price
+
+    @property
+    def shares(self):
+        return self._shares
+
+    @shares.setter
+    def shares(self, value):
+        if not isinstance(value, int):
+            raise TypeError('Expected an integer')
+        self._shares = value
+
+    def sell(self, number):
+        self.shares -= number
+
+    def __repr__(self):
+        return f'Stock({self.name}, {self.shares}, {self.price})'
+
+
+class MyStock(Stock):
+    def __init__(self, name, shares, price, factor):
+        # Check the call to `super` and `__init__`
+        super().__init__(name, shares, price)
+        self.factor = factor
+
+    def panic(self):
+        self.sell(self.shares)
+
+    @property
+    def cost(self):
+        return self.factor * super().cost
diff --git a/Work/tableformat.py b/Work/tableformat.py
@@ -0,0 +1,80 @@
+# tableformat.py
+
+class TableFormatter:
+    def headings(self, headers):
+        """
+        Emit the table headings.
+        """
+        raise NotImplementedError()
+
+    def row(self, rowdata):
+        """
+        Emit a single row of table data.
+        """
+        raise NotImplementedError()
+
+
+class TextTableFormatter(TableFormatter):
+    """
+    Emit a table in plain-text format
+    """
+    def headings(self, headers):
+        for h in headers:
+            print(f'{h:>10s}', end=' ')
+        print()
+        print(('-' * 10 + ' ') * len(headers))
+
+    def row(self, row_data):
+        for column in row_data:
+            print(f'{column:>10}', end=' ')
+        print()
+
+
+class CSVTableFormatter(TableFormatter):
+    """
+    Output portfolio data in CSV format.
+    """
+    def headings(self, headers):
+        print(','.join(headers))
+
+    def row(self, row_data):
+        print(','.join(row_data))
+
+class HTMLTableFormatter(TableFormatter):
+    """
+    Output portfolio data in HTML format.
+    """
+    def print_row(self, data, wrapper='td'):
+        print('<tr>', end='')
+        for column in data:
+            print(f'<{wrapper}>{column}</{wrapper}>', end='')
+        print('</tr>')
+
+    def headings(self, headers):
+        self.print_row(headers, 'th')
+
+    def row(self, row_data):
+        self.print_row(row_data)
+
+
+def create_formatter(name):
+    if name == 'txt':
+        formatter = TextTableFormatter()
+    elif name == 'csv':
+        formatter = CSVTableFormatter()
+    elif name == 'html':
+        formatter = HTMLTableFormatter()
+    else:
+        raise FormatError(f'Unknown table format {name}')
+    return formatter
+
+
+def print_table(portfolio, columns, formatter):
+    formatter.headings(columns)
+    for holding in portfolio:
+        line = [getattr(holding, column) for column in columns]
+        formatter.row(line)
+
+
+class FormatError(Exception):
+    pass
