Overview
I've just written a package in Python (which I've named sibstructures
) whose purpose is to assist staff in managing a coldstore, which is a warehouse-sized refrigerator holding (in this case) boxes of potatoes.
The code below is one file taken from that package, the file which defines the centrepiece Coldstore
class. It also defines a number of "helper" classes, i.e. small classes whose sole purpose is to facilitate something or other in Coldstore
. The file then finishes with some unit tests.
I'm pretty much self-taught Python, so any remarks would be greatly appreciated, but I'd be particularly grateful for any guidance - or praise - regarding the way in which I've structured the file.
The Code
Opening Lines
This follows the proforma by which I begin almost all of my .py files.
### This code defines a class, which models a potato coldstore.
# Imports.
import time
# Custom imports.
from sibstructures import sibconstants
from sibstructures import data_storage
from sibstructures import layer_colours
from sibstructures.box import Box
from sibstructures.spot import Spot
from sibstructures.numerals import to_roman, to_arabic
# Local constants.
max_tickets = 100
Main Class
There then follows the definition of the "main" class, which in this case models the coldstore itself.
A few features are worth explaining:
- The
data_storage
module allows access to a database, which (1) records what enters and exits the coldstore, (2) records where boxes are placed, and (3) stores plans for where boxes should be stored. - Scanners (using RFID) identify boxes as they enter and leave.
- Using these scanners, the coldstore issues a recommendation to the storesman as to where to put a given box as it enters.
- Upon receiving a recommendation, the storesman may either (1) follow the recommendation, (2) cancel the recommendation, or (3) override the recommendation (i.e. put the box somewhere else).
##############
# MAIN CLASS #
##############
# The class in question.
class Coldstore:
def __init__(self, code):
self.code = code
self.columns = None
self.headlands = None
self.ticket_machine = Ticket_Machine()
self.outstanding_recommendations = []
# Load a map of this coldstore (when empty) from plain text.
def load_floorplan_from_plain_text(self, plain_text):
lines = plain_text.split("\n")
no_of_columns = len(lines[0].split(" "))
sheet_columns = []
for i in range(no_of_columns):
max_layers_list = []
for line in lines:
max_layers = line.split(" ")
max_layers_list.append(int(max_layers[i]))
max_layers_tuple = tuple(max_layers_list)
sheet_columns.append(max_layers_tuple)
result = []
i = 0
for item in sheet_columns:
letter = chr(ord('a')+i)
column = Column(letter, item)
result.append(column)
self.columns = result
# Load the headlands of this coldstore from two integers.
def load_headlands_from_arguments(self, no_of_spots, max_layers):
result = []
for i in range(1, no_of_spots+1):
numeral = to_roman(i)
spot = Spot(numeral, max_layers)
result.append(spot)
self.headlands = result
# Load a coldstore's floorplan from the database.
def load_floorplan_from_db(self):
data = data_storage.fetch_local_specifics()
if data == None:
raise Exception("No local specifics to load.")
else:
floorplan = data["floorplan"]
headlands_spots = data["headlands_spots"]
headlands_maxlayers = data["headlands_maxlayers"]
self.load_floorplan_from_plain_text(floorplan)
self.load_headlands_from_arguments(headlands_spots,
headlands_maxlayers)
# Ronseal.
def select_column_by_letter(self, letter):
for column in self.columns:
if column.letter == letter:
return column
raise Exception("No column with letter="+letter+".")
# Executes the delivery of a box to a given column, and logs the same.
def execute_drop(self, epc, column_letter, row_numeral):
box = Box(epc, None)
column = self.select_column_by_letter(column_letter)
spot = column.select_spot_by_row(row_numeral)
if spot.add_one(box):
data_storage.commit_drop(epc, column_letter, spot.row_id,
int(time.time()))
else:
raise Exception("It should be impossible to add a box at column="+
column_letter+", row="+row_numeral+".")
# Take action upon detecting a given EPC entering this coldstore.
def execute_entry(self, epc, timestamp):
data_storage.add_entry_log(epc, timestamp)
self.issue_recommendation(epc)
# Take action upon detecting a given EPC leaving this coldstore.
def execute_exit(self, epc, timestamp):
data_storage.add_exit_log(epc, timestamp)
result = self.remove_box_by_epc(epc)
return result
# Add a recommendation to the list, given that a specific EPC was detected
# entering this coldstore.
def issue_recommendation(self, epc):
ticket = self.ticket_machine.issue_ticket()
permitted_columns = data_storage.fetch_permitted_columns(epc)
for letter in permitted_columns:
column = self.select_column_by_letter(letter)
for spot in column.spots:
if spot.has_room():
column_letter = letter
row_numeral = spot.row_id
recommendation = Recommendation(ticket, epc, column_letter,
row_numeral)
self.outstanding_recommendations.append(recommendation)
return
column_letter = None
row_numeral = None
recommendation = Recommendation(ticket, epc, column_letter, row_numeral)
self.outstanding_recommendations.append(recommendation)
# Remove a box with a given EPC from this coldstore, if possible.
def remove_box_by_epc(self, epc):
for column in self.columns:
for spot in column.spots:
if spot.remove_one(epc):
return True
return False
# Select a recommendation by its ticket.
def select_recommendation_by_ticket(self, ticket):
for rec in self.outstanding_recommendations:
if rec.ticket == ticket:
return rec
return None
# Follow up on an outstanding recommendation.
def follow_recommendation(self, ticket):
rec = self.select_recommendation_by_ticket(ticket)
if rec == None:
raise Exception("No outstanding recommendation with ticket="+
str(ticket)+".")
elif rec.column_letter == None:
raise Exception("Cannot follow a void recommendation.")
else:
self.execute_drop(rec.epc, rec.column_letter, rec.row_numeral)
self.outstanding_recommendations.remove(rec)
self.ticket_machine.return_ticket(ticket)
# Cancel a given outstanding recommendation.
def cancel_recommendation(self, ticket):
rec = self.select_recommendation_by_ticket(ticket)
self.outstanding_recommendations.remove(rec)
self.ticket_machine.return_ticket(ticket)
# Cancel a given outstanding recommendation, and do something else with
# that box.
def override_recommendation(self, ticket, column_letter, row_numeral):
rec = self.select_recommendation_by_ticket(ticket)
epc = rec.epc
self.cancel_recommendation(ticket)
self.execute_drop(epc, column_letter, row_numeral)
self.ticket_machine.return_ticket(ticket)
# Ronseal.
def print_me(self):
printer = Coldstore_Printer(self)
printer.print_me()
Helper Classes
A chocolate box of small, mostly self-explanatory classes.
################################
# HELPER CLASSES AND FUNCTIONS #
################################
# A class which models a single column of a coldstore.
class Column:
def __init__(self, letter, max_layers_tuple):
self.letter = letter
self.spots = self.make_spots(max_layers_tuple)
# Add the column's spots, from a tuple of their max_layers.
def make_spots(self, max_layers_tuple):
i = 1
result = []
for max_layers in max_layers_tuple:
numeral = to_roman(i)
spot = Spot(numeral, max_layers)
result.append(spot)
i = i+1
return result
# Ronseal.
def select_spot_by_row(self, row_id):
for spot in self.spots:
if spot.row_id == row_id:
return spot
raise Exception("No spot with row_id="+row_id+".")
# A class which issues and recycles tickets, used to keep track of any
# recommendations issued.
class Ticket_Machine:
def __init__(self):
self.tickets = set()
for i in range(max_tickets):
ticket = i+1
self.tickets.add(ticket)
# Ronseal.
def issue_ticket(self):
result = self.tickets.pop()
return result
# Ronseal.
def return_ticket(self, ticket):
self.tickets.add(ticket)
# A bare-bones class to hold the properties of a recommmendation.
class Recommendation:
def __init__(self, ticket, epc, column_letter, row_numeral):
self.ticket = ticket
self.epc = epc
self.column_letter = column_letter
self.row_numeral = row_numeral
# A class which permits the printing of a coldstore to the screen.
class Coldstore_Printer:
def __init__(self, coldstore):
self.coldstore = coldstore
self.rows = self.make_rows()
# Rearrange the coldstore's columns into rows, which can be printed more
# easily, and extract the useful data from each spot.
def make_rows(self):
result = []
for i in range(len(self.coldstore.columns[0].spots)):
row = []
for column in self.coldstore.columns:
printed_spot = Printed_Spot(column.spots[i])
row.append(printed_spot)
result.append(row)
return result
# Makes the top row of the printout.
def get_top_row(self):
result = " "
for i in range(len(self.rows[0])):
result = result+" "+chr(ord('a')+i)+" "
return result
# Makes the tween rows of the printout.
def get_tween_row(self):
result = "+"
for _ in range(len(self.rows[0])):
result = result+"-----+"
return result
# Ronseal.
def print_me(self):
result = []
top_row = self.get_top_row()
tween_row = self.get_tween_row()
i = 1
result.append(top_row)
result.append(tween_row)
for row in self.rows:
printout_for_row = "|"
for spot in row:
printout_for_row = printout_for_row+" "+spot.get_printout()+" |"
printout_for_row = printout_for_row+" "+to_roman(i)
result.append(printout_for_row)
result.append(tween_row)
i = i+1
for item in result:
print(item)
# A class which is a component of the Coldstore_Printer class.
class Printed_Spot:
LAYER_COLOURS = (None, layer_colours.RED, layer_colours.ORANGE,
layer_colours.YELLOW, layer_colours.GREEN,
layer_colours.BLUE, layer_colours.INDIGO,
layer_colours.VIOLET)
MAXED_OUT_LAYER_COLOUR = layer_colours.VIOLET
RESET = layer_colours.RESET
def __init__(self, spot):
self.layers = spot.layers()
if spot.max_layers == 0:
self.is_void = True
else:
self.is_void = False
if self.layers == 0:
self.is_empty = True
else:
self.is_empty = False
if (self.is_void or self.is_empty):
self.top_display_code = None
else:
self.top_display_code = spot.boxes[-1].display_code
# Ronseal.
def get_printout(self):
if self.is_void:
result = "XXX"
elif self.is_empty:
result = " "
else:
colour = Printed_Spot.LAYER_COLOURS[self.layers]
result = colour+self.top_display_code+Printed_Spot.RESET
return result
Testing
Finally, there's some unit-testing. It's mostly unambitious: I try to make sure that the classes do what they're supposed to do, and don't try too hard to break them.
###########
# TESTING #
###########
# Test the Column class.
def test_column():
column = Column('a', (0, 2, 3))
assert(column.letter == 'a')
assert(column.spots[0].max_layers == 0)
assert(column.spots[0].row_id == "i")
assert(column.spots[1].max_layers == 2)
assert(column.spots[1].row_id == "ii")
assert(column.spots[2].max_layers == 3)
assert(column.spots[2].row_id == "iii")
assert(len(column.spots[2].boxes) == 0)
# Test the Coldstore class.
def test_coldstore():
test_floorplans()
# These tests depend on agreeable data in the DB.
test_drops()
test_entry_and_exit()
test_recommendations()
#test_printing()
# Test the loading of floorplans by various means.
def test_floorplans():
plain_text = ("1 2 3\n"+
"4 5 6\n"+
"7 8 9")
coldstore = Coldstore("cs0")
coldstore.load_floorplan_from_plain_text(plain_text)
assert(len(coldstore.columns) == 3)
column_a = coldstore.columns[0]
assert(column_a.letter == 'a')
assert(len(column_a.spots) == 3)
spot_i = column_a.spots[0]
assert(spot_i.row_id == "i")
assert(spot_i.max_layers == 1)
spot_iii = column_a.spots[2]
assert(spot_iii.row_id == "iii")
assert(spot_iii.max_layers == 7)
# The remaining tests depend on agreeable data in the DB.
coldstore.load_floorplan_from_db()
assert(len(coldstore.columns) == 3)
column_a = coldstore.columns[0]
assert(column_a.letter == 'a')
assert(len(column_a.spots) == 3)
spot_i = column_a.spots[0]
assert(spot_i.row_id == "i")
assert(spot_i.max_layers == 1)
spot_iii = column_a.spots[2]
assert(spot_iii.row_id == "iii")
assert(spot_iii.max_layers == 7)
assert(len(coldstore.headlands) == 3)
assert(coldstore.headlands[0].max_layers == 7)
assert(coldstore.headlands[2].max_layers == 7)
# Test dropping boxes. Depends on agreeable data in DB.
def test_drops():
coldstore = Coldstore("cs0")
coldstore.load_floorplan_from_db()
coldstore.execute_drop("epc0", 'a', "i")
# Now check database.
# Test the entry and exit routines. Depends on agreeable data in DB.
def test_entry_and_exit():
coldstore = Coldstore("cs0")
coldstore.load_floorplan_from_db()
coldstore.execute_entry("epc0", int(time.time()))
assert(coldstore.execute_exit("epc0", int(time.time())) == False)
rec = coldstore.outstanding_recommendations[0]
assert(rec.epc == "epc0")
assert(rec.column_letter == 'a')
assert(rec.row_numeral == "i")
# Test the recommendation system. Depends on agreeable data in DB.
def test_recommendations():
coldstore = Coldstore("cs0")
coldstore.load_floorplan_from_db()
coldstore.execute_entry("epc0", int(time.time()))
assert(len(coldstore.ticket_machine.tickets) == max_tickets-1)
coldstore.follow_recommendation(
coldstore.outstanding_recommendations[0].ticket)
assert(coldstore.execute_exit("epc0", int(time.time())) == True)
assert(len(coldstore.ticket_machine.tickets) == max_tickets)
coldstore.execute_entry("epc0", int(time.time()))
coldstore.cancel_recommendation(
coldstore.outstanding_recommendations[0].ticket)
assert(len(coldstore.outstanding_recommendations) == 0)
coldstore.execute_entry("epc0", int(time.time()))
coldstore.override_recommendation(
coldstore.outstanding_recommendations[0].ticket, 'a', "i")
assert(len(coldstore.outstanding_recommendations) == 0)
assert(coldstore.execute_exit("epc0", int(time.time())) == True)
# A sub-function for "test_printing()".
def introduce_epc(coldstore, epc):
if len(coldstore.outstanding_recommendations) != 0:
raise Exception("Pre-existing outstanding recommendations.")
coldstore.execute_entry(epc, int(time.time()))
coldstore.follow_recommendation(
coldstore.outstanding_recommendations[0].ticket)
# Test the facilities used to print coldstores to the screen.
def test_printing():
coldstore = Coldstore("cs0")
coldstore.load_floorplan_from_db()
introduce_epc(coldstore, "epc0")
introduce_epc(coldstore, "epc1")
introduce_epc(coldstore, "epc2")
introduce_epc(coldstore, "epc3")
introduce_epc(coldstore, "epc4")
introduce_epc(coldstore, "epc5")
coldstore.print_me()
# Run the unit tests.
def test():
test_column()
test_coldstore()
print("Tests passed!")
Wrapping Up
I always put this at the end of my .py files, so I thought I'd check I've been doing the right thing!
###################
# RUN AND WRAP UP #
###################
def run():
test()
if __name__ == "__main__":
run()
Other Stuff
A "review copy" of the whole package is available at https://github.com/tomhosker/sibstructures4review.
python sibstructures
,cd sibstructures; python main.py
,python -m sibstructures
orpytest
to run your tests? \$\endgroup\$ – Peilonrayz Jan 22 '20 at 18:41sibstructures
in multiple, disparate places in the repository I'm working on, and so I concluded that the best solution would be to makesibstructures
into a custom package by copying it into~/.local/lib/python3.6
- that path being specific to Linux. (If you know of a better solution, I'd very much like to learn about it.) I can then run this file - namedcoldstore.py
- by navigating into the appropriate directory and runningpython3 coldstore.py
, and that's it. \$\endgroup\$ – Tom Hosker Jan 22 '20 at 22:58