Justin Frank / neurocube_consumer

Commit 6556b8b2276467877ef733dbb560cb24bcf305a6

Authored by Justin Frank 2021-06-28 19:55:45 -0700

Exists in master

added more docstrings

Showing 1 changed file with 70 additions and 23 deletions Side-by-side Diff

utils/consume.py

...	...	@@ -63,47 +63,87 @@
63	63	"""
64	64	currentRow = 0
65	65	queue = MyQueue()
66		- currentDf = pd.DataFrame()
67	66
68		- def __init__(self, row, queue, df):
	67	+ def __init__(self, row, queue):
	68	+ """Initialize the myCsv handler.
	69	+
	70	+ Parameters
	71	+ ----------
	72	+ row : int or list of int
	73	+ The index of the first row to begin processing at
	74	+ queue : MyQueue
	75	+ The queue in which to store row indexes in
	76	+ """
69	77	self.currentRow = row
70	78	self.queue = queue
71		- self.currentDf = df
72	79
73	80	def on_created(self, event):
	81	+ """Prints the file path if the file being watched for is created.
	82	+
	83	+ Parameters
	84	+ ----------
	85	+ event : event
	86	+ The event object representing the event that triggered the handler.
	87	+ """
74	88	print("Watchdog received created event - % s." % event.src_path)
75	89	# Event is created, you can process it now
76	90
77	91	def on_modified(self, event):
	92	+ """Prints the file path if the file being watched for is modified and adds the indecies of all
	93	+ rows in the csv including and after "self.currentRow".
	94	+
	95	+ Parameters
	96	+ ----------
	97	+ event : event
	98	+ The event object representing the event that triggered the handler.
	99	+ """
78	100	print("Watchdog received modified event - % s." % event.src_path)
79		- self.currentDf = pd.read_csv(event.src_path)
80		- lastRow = self.currentDf.shape[0]
81		- #TODO: generate a list of row numbers from self.current row+1 to the last row in the df
	101	+ currentDf = pd.read_csv(event.src_path)
	102	+ lastRow = currentDf.shape[0]
82	103	rowSpan = get_row_span(self.currentRow, lastRow)
83		- #TODO: put this list of numbers in the queue
84		- print('rowspan: ')
85		- print(rowSpan)
	104	+ #print('rowspan: ')
	105	+ #print(rowSpan)
86	106	self.queue.enqueue(rowSpan)
87		- #TODO: update self.current row
88	107	self.currentRow = lastRow
89	108	# Event is modified, youu\ can process it now
90	109
91		-class MyZipHandler(FileSystemEventHandler):
92		- queue = []
	110	+# class MyZipHandler(FileSystemEventHandler):
	111	+# queue = []
93	112
94		- def __init__(self, row, queue):
95		- self.queue = queue
	113	+# def __init__(self, queue):
	114	+# """Initialize the MyZipHandler.
96	115
97		- def on_created(self, event):
98		- print("Watchdog received created event - % s." % event.src_path)
99		- # Event is created, you can process it now
	116	+# Parameters
	117	+# ----------
	118	+# queue : MyQueue
	119	+# The queue in which to store row indexes in
	120	+# """
	121	+# self.queue = queue
100	122
101		- def on_modified(self, event):
102		- print("Watchdog received modified event - % s." % event.src_path)
103		- self.queue.append(event.src_path)
104		- # Event is modified, youu\ can process it now
	123	+# def on_created(self, event):
	124	+# """Prints the file path if the file being watched for is created and adds the indecies of all.
105	125
	126	+# Parameters
	127	+# ----------
	128	+# event : event
	129	+# The event object representing the event that triggered the handler.
	130	+# """
	131	+# print("Watchdog received created event - % s." % event.src_path)
	132	+# # Event is created, you can process it now
106	133
	134	+# def on_modified(self, event):
	135	+# """Prints the file path if the file being watched is modified and adds the path to a queue.
	136	+
	137	+# Parameters
	138	+# ----------
	139	+# event : event
	140	+# The event object representing the event that triggered the handler.
	141	+# """
	142	+# print("Watchdog received modified event - % s." % event.src_path)
	143	+# self.queue.append(event.src_path)
	144	+# # Event is modified, youu\ can process it now
	145	+
	146	+
107	147	def consume(path,line):
108	148	"""Processes neurocube jobs by watching the gdfuse file system.
109	149
110	150
...	...	@@ -120,13 +160,13 @@
120	160	"""
121	161	queue = MyQueue()
122	162	df = pd.DataFrame()
123		- event_handler = MyCSVHandler(line, queue, df)
	163	+ event_handler = MyCSVHandler(line, queue)
124	164	observer = Observer()
125	165	observer.schedule(event_handler, path, recursive=True)
126	166	observer.start()
127	167	try:
128	168	while True:
129		- time.sleep(1)
	169	+ time.sleep(1) #This may not be needed
130	170	currentJob = queue.dequeue()
131	171	if currentJob != None:
132	172	print(df)
...	...	@@ -169,5 +209,12 @@
169	209	print(row)
170	210
171	211	def get_row(job,df):
	212	+ """Prints a row from the csv file.
	213	+
	214	+ Parameters
	215	+ ----------
	216	+ mountPoint : str
	217	+ The directory to mount the google drive to
	218	+ """
172	219	return df.iloc[job]