In my Python journey I was looking for a "real life" problem and I think I have found a good one at my current work.
The laser CNC machine I'm working with generates reports that contains timings of how the laser was used throughout the day, which metal parts were cut, how long did it take and how much time laser was idle.
Reports are being generated daily. If I was a manager I would be interested in these reports, but in their HTML table format is not very useful, as you can see in these examples in my GitHub project.
Because this is more of a learning task I decided to code it as a tool (?) that provides parsed data so the user can use parsed data as he/she wants.
When coding this project I came up with 3 classes. Each has its own python module:
This is the Parser class
from datetime import timedelta
from collections import defaultdict
import lxml.html as p
class ReportParser:
"""Parser class
Generally used just to encapsulate methods that parse html report and
process data.
"""
def parse(self, file):
self.file = file
self.tree = self.get_etree()
self.raw_data = self.get_raw_data()
self.prepare_data()
return self.process_data()
def get_etree(self):
"""Returns root of the parsed document"""
return p.parse(self.file).getroot()
def get_raw_data(self):
"""Returns list of dicts with parsed data.
Method that parse data from html document and wrap it in handy for
futher proccesing format.
Returns
-------
list
parsed data wrapped in list
"""
data = []
for elem in self.tree.iter('tr'):
# will not work in python 2
time, name, status, *_ = elem
data.append([time.text, name.text, status.text])
if not data:
raise TypeError('''Report data is empty, make sure report has data
wrapped in <tr> or <TR> tags''')
return data[1:] # cut the header tags <tr><th></th></tr>
@staticmethod
def prepare_time(time):
"""Returns timedelta object converted from string
Parameters
----------
time : str
Time string
Returns
-------
timedelta
timedelta object converted from string
Example
-------
>>> prepare_time('10:11:12')
datetime.timedelta(0, 36672)
"""
t_string = time.split(':')
return timedelta(hours=int(t_string[0]),
minutes=int(t_string[1]),
seconds=int(t_string[2])
)
@staticmethod
def prepare_name(name):
"""Returns program name without leading path and program version
Parameters
----------
name : str
Program name
Returns
-------
str
Simplified program name
Example
-------
>>> prepare_name('1/2/3/4/some_NAMEver23.04.ISO')
some_NAME.ISO
"""
name = name.split('/')[-1]
if 'ver' in name:
name = name.split('ver')[0] + '.ISO'
return name
def _convert_time(self):
"""Converts all time string in data to timedelta objects"""
for data_row in self.raw_data:
data_row[0] = self.prepare_time(data_row[0])
def _convert_name(self):
"""Converts all program names in data to simplified representation"""
for data_row in self.raw_data:
data_row[1] = self.prepare_name(data_row[1])
def prepare_data(self):
self._convert_time()
self._convert_name()
def process_data(self):
"""Summarize parsed data
Explanation of algorithm:
+----------+------+---------+
| Time | Name | Status |
+----------+------+---------+
| 00:00:00 | prg1 | STARTED |
| 01:00:00 | prg1 | STOPPED |
| 01:05:00 | prg2 | STARTED |
| 02:05:00 | prg2 | STOPPED |
| 02:10:00 | prg3 | STARTED |
| 03:10:00 | prg3 | STOPPED |
+----------+------+---------+
Each report reprsent the working day.
Time in first column is timedelta objects.
At each iteration we need to determine how much time the program was in
work or laser was off work, time interval between two statuses reprsent
this time, whether program was in work (started->stopped) or laser
was in idle (stopped->started).
'current' - current time in "table", changes at each iteration.
1st row. prg1 started, calculate how much time passed from previous
stopped status (idle): 00:00:00 - current (which is 00:00:00),
new current is 1st row "time" (00:00:00).
2nd row. prg1 stopped, calculate how much time passed from started
status: 01:00:00 - current, writes data to dict: {'prg1':01:00:00},
New current is 2nd row "time" (01:00:00).
Repeating the procces above until StopIteration raised we'll get:
data = {'prg1': 01:00:00, 'prg2': 01:00:00, 'prg3': 01:00:00},
current = 03:10:00 and idle equals to 00:10:00.
As StopIteration raised we need to know which status was last, in case
above it is "STOPPED", so we need to calculate how much time passed
from the time in last row to the end of the day:
24:00:00 - 03:10:00, idle + 20:50:00 which sums up to a total 21:00:00
now: data = {'prg1': 01:00:00, 'prg2': 01:00:00, 'prg3': 01:00:00,
'idle': 21:00:00}
In case where last status is STARTED we need to calculate how much time
last program was in work till the day end and add it to results.
"""
idle = timedelta() # collects ammout of time laser wasn't working
current = timedelta() # used to determine current position in table
data = defaultdict(timedelta)
# Using manual iteration because we need to handle the end of iteration
# and because I don't want to use for/else thing.
i = iter(self.raw_data)
try:
while True:
data_row = next(i) # look like that: [00:00:00, prg1, STARTED]
time, name, status = data_row
if status == 'STARTED':
idle += time - current
current = time
else:
data[name] += time - current
current = time
except StopIteration:
if status == 'STARTED':
data[name] += timedelta(days=1) - current
else:
idle += timedelta(days=1) - current
data['idle'] += idle
return data
This is the Report class that represents the HTML report:
import os
from .utils import get_date
class Report:
"""
A simple report class that store data related to report.
Reports with the same date counts as equal.
"""
def __init__(self, path, data=None):
self.path = path
self.name = os.path.split(self.path)[1]
self.date = get_date(self.name)
self.data = data
@property
def programs(self):
"""Returns parsed programs for a report"""
return self.data.keys()
@property
def timings(self):
"""Returns parsed timings for a report"""
return self.data.values()
def items(self):
return self.data.items()
def __eq__(self, other):
return self.date == other
def __gt__(self, other):
return self.date > other
def __lt__(self, other):
return self.date < other
def __ge__(self, other):
return self.__gt__(other) or self.__eq__(other)
def __le__(self, other):
return self.__lt__(other) or self.__eq__(other)
def __hash__(self):
return hash(self.date)
def __repr__(self):
return "Report({}, {})".format(self.name,
self.date.strftime('%Y_%m_%d'))
This is Storage class which is used to store the Reports and do some job on sequences of Reports:
import pickle
from collections import defaultdict, Sequence
from datetime import timedelta
from .report import Report
from .utils import update_default_dict, convert_date
class Storage:
"""Storage class"""
def __init__(self, path=None, date_format='%Y_%m_%d'):
self.date_format = date_format
self.content = set()
if path:
self.path = path
try:
self.load(self.path)
except FileNotFoundError:
pass
else:
self.path = 'datadump.pickle'
@property
def reports_names(self):
"""Storage property. Returns names of reports stored in storage"""
return {x.name for x in self.content}
def save(self, path=None):
"""Serialize storage content to a pickle file
Parameters
----------
path : str
Path where pickle file should be saved. If None using self.path
"""
if path is None:
path = self.path
with open(path, 'wb') as f:
pickle.dump(self.content, f)
def load(self, path, override=False):
"""Deserialize pickle file to a python object
Parameters
----------
path : str
Path to a pickle file that should be loaded
override: bool
Delete current storage content if True
"""
if override:
self.content = set()
with open(path, 'rb') as f:
self.add(pickle.load(f))
def get_by_date(self, date):
"""Returns report object from storage content according to passed date
Parameters
----------
date : str | datetime
Interesting report's date.
Returns
-------
Report
Report object with .date == date
None
If there is no Report objects with .date == date in self.content
Examples
--------
>>> get_by_date('2017_04_04')
Report(2017_04_04.html, 2017_04_04)
>>> get_by_date('2032_04_07')
None
"""
item = self.filter_by_date(self.content, date, date)
return item.pop() if item else None
def filter_by_date(self, data, _max, _min):
"""Returns Report objects in _max, _min date range
Parameters
----------
data: sequence
Sequence with Report objects to filter
_max : str | datetime
maximal date limit to filter
_min : str | datetime
minimal date limit to filter
Returns
-------
set
Set with report objects in _max, _min date range
Empty if there is no reports in given range in self.content
"""
_max, _min = [convert_date(x, self.date_format) for x in (_max, _min)]
return {x for x in data if _max >= x >= _min}
def add(self, item):
"""Adds item to storage.content
You can use this method to add a particular Report or a
sequence of reports.
Parameters
----------
item : Report | sequence
item to add to the content
"""
if isinstance(item, str):
raise TypeError("""Not supported type.
Make sure passed sequence is
one of the following types:
list, tuple or set.""")
# We need to use two different methods depending on if item is
# unhashable or iterable.
elif isinstance(item, Sequence) or isinstance(item, set):
self._add_sequence(item)
elif isinstance(item, Report):
# used for non iterable, hashable item.
self._add_item(item)
else:
raise TypeError("""Not supported type.
Make sure passed item is either Report instance
or a sequence""")
def _add_sequence(self, seq):
"""Adds sequence to self.content
Adds sequence only if every item in this sequence is Report instance
Parameters
----------
seq : sequence
sequence to add
"""
if all(isinstance(x, Report) for x in seq):
self.content.update(seq)
else:
raise TypeError('Sequence contains not instances of Report class')
def _add_item(self, item):
"""Adds item to self.content
Adds item only if items is Report instance
Parameters
----------
item : Report
Report to add
"""
self.content.add(item)
def delete(self, date):
"""Deletes report from self.content
Use this method to delete report with .date == date
Parameters
----------
date : str | datetime
Report's date
"""
item = self.get_by_date(date)
if item is not None:
self.content.remove(item)
else:
print('No {} report in content'.format(date.strftime('%Y_%m_%d')))
@staticmethod
def dump_data(to_dump):
"""Returns summarized data of all passed reports
Parameters
----------
to_dump : sequence
Sequence that contains reports
Returns
-------
data : defaultdict
defauldict with summarized data of all passed reports
"""
data = defaultdict(timedelta)
for item in to_dump:
update_default_dict(data, item)
return data
And there is some utility functions:
from datetime import datetime, timedelta
def convert_timedelta(item):
"""Returns formated timedelta string representation
Function used to covnert timedelta string representation form this:
'1 day, 10:11:12' to this: '34h 11m 12s'.
Parameters:
-----------
item : timedelta
timedelta object that should be converted
Returns:
--------
str
timedelta string representation
"""
if isinstance(item, timedelta):
seconds = int(item.total_seconds())
hours, remainder = divmod(seconds, 3600)
minutes, seconds = divmod(remainder, 60)
formated = '{}h {}m {}s'.format(hours, minutes, seconds)
else:
raise ValueError(item, 'is not timedelta object')
return formated
def convert_date(date_string, s_format='%Y_%m_%d'):
"""Returns datetime object
Function used to convert string to datetime object.
If date_string is already a datetime object just returns it.
Parameters
----------
date_string : string | datetime
string or datetime object
s_format : string
format of the date_string
Returns
-------
datetime
datetime object converted or not converted from date_string
"""
if isinstance(date_string, str):
return datetime.strptime(date_string, s_format)
elif isinstance(date_string, datetime):
return date_string
else:
raise ValueError(date_string, 'is not a string or datetime object')
def update_default_dict(main, other):
"""Summarize values of two defaultdicts
Function sumarize two defaultdicts values if they both have similar
keys and just adds key and value if don't
Parameters
----------
main : defaultdict
dict that should be updated
other : defaultdict
dict whih keys and values will be used to update main defaultdict
Returns
-------
None
"""
for k, v in other.items():
main[k] += v
def get_date(name):
"""Returns convert_date() function with extracted date string as argument
Function extract date string from a report name, for example:
get_date('2017_07_04.html') will return convert_date('2017_07_04')
Parameters
----------
name : str
report name
Returns
-------
datetime
the result of calling convert_date() function with name as argument
"""
return convert_date(name.split('.')[0])
And here how you can use it:
import os
import sys
from ppowerparser.parse import ReportParser
from ppowerparser.report import Report
from ppowerparser.storage import Storage
folder = r'C:\some\path\to\folder\with\reports'
# collect *.html files in folder
files = [x for x in os.listdir(folder) if x.endswith('.html')]
parser = ReportParser() # creates parser itself
storage = Storage() # storage to store parsed data
# Parse each valid html report in folder, create Report object for
# each file and adds them to the storage.
for item in files:
path = os.path.join(folder, item)
data = parser.parse(path)
storage.add(Report(path, data))
storage.save() # serialize set of Report objects to a pickle file
# Summarize data from Report objects
data = storage.dump_data(storage.content)
# Sort data in descending order.
sorted_tuples = sorted(data.items(), key=lambda x: x[1],
reverse=True)
print('\nSummarized data: \n')
for k, v in sorted_tuples:
print('{} - {}'.format(k, convert_timedelta(v)))
Which will output something like this:
Summarized data:
push_bezLogo_S3_19.8.ISO - 34h 20m 25s
idle - 26h 53m 44s
ProfilKrepleniyGruza_S2.5.ISO - 16h 53m 52s
745461.001_01_02.ISO - 15h 30m 55s
745322.010_VER23.7_S3.ISO - 10h 49m 18s
745312.012_S4_.ISO - 8h 51m 46s
6306026_50_S4.ISO - 7h 45m 21s
...
VPO.ISO - 0h 3m 5s
745275.003_5.3.ISO - 0h 2m 24s
FOCAL_TEST_NEW.ISO - 0h 1m 24s
735636.004_S2_5.3.ISO - 0h 0m 34s
741121.001_S3.ISO - 0h 0m 29s
batman-logo-bw.ISO - 0h 0m 28s
I know there is a lot of code in my question and I would appreciate so much if someone could go through it and tell me about what he or she thinks, and maybe answer questions about a couple things that bother me:
Is this a normal thing to have 3 modules for 3 different classes? As I can judge it helps to separate/organize code, but also I've heard that a good python package should't contain a lot of modules (famous youtube video 'Stop writing classes') and classes doing code separation very good, so maybe I should use 1 module that will contain all 3 classes?
Is the ReportParser class even necessary? Maybe best approach to this kind of problem is to just place bunch of functions in a single file?
Do you find Storage class confusing and unnecessary?
repo now contains refactored code 05.11.2017 If it's easier to go through the code in an IDE/locally, here is a link to the repository on github.
