holgi
/
arduino-timetable
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity
Browse Source

some cleanup

pyinstaller
Holger Frey 7 years ago
parent
da93ef2f74
commit
dd6e3493b9
3 changed files with 360 additions and 331 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 342
      arduino_timetable/__init__.py
  2. 344
      arduino_timetable/timetable.py
  3. 5
      run_magnetic_trap.py

342
arduino_timetable/__init__.py
Unescape View File

@ -1,333 +1,13 @@
''' Controlling an Arduino with a time table ''' ''' Controlling an Arduino with commands in a time table '''
import asyncio from .gui import run_application
import serial from .timetable import (
import serial.tools.list_ports parse_time_table,
parse_excel_file,
from collections import namedtuple find_arduino_port,
from datetime import datetime, timedelta send_command,
from openpyxl import load_workbook run,
TimedCommands
)
__version__ = '0.0.1' __version__ = '0.0.1'
# example time table for the TrapControl
example = '''
00:00.0 close
00:06.0 open
00:08.0 close
00:10.0 open
00:12.0 close
00:14.0 open
00:16.0 close
00:18.0 open
00:20.0 close
00:22.0 open
'''
# datetime formats for timetable parsing
TIMETABLE_FORMATS = ['%M:%S', '%M:%S.%f', '%H:%M:%S', '%H:%M:%S.%f']
# named tuple for a single scheduled command
ScheduledCommand = namedtuple('ScheduledCommand', 'delta command cmd')
def parse_time_table(timetable, available_commands):
''' parses a (text) time table into a list of ScheduledCommands
timetable:
a textual representaion of time and commands like
00:12.0 close
00:14.0 open
00:16.0 close
00:18.0 open
available_commands:
a dictionary containing the available human readable commands as
keys and the commands to send as values
{'open': 0, 'close': 1}
returns a list consisting of ScheduledCommands
'''
# split the time table text by line break
raw_lines = timetable.split('\n')
# remove surrounding white space
content_lines = (line.strip() for line in raw_lines)
# remove empty lines
lines = (line for line in content_lines if line)
timed_commands =[]
for line in lines:
# split the lines into time and command
try:
raw_time, raw_command = line.split(None, 1)
except ValueError:
msg = "error in line '{}'".format(line)
raise ValueError(msg)
# parse time and commands
delta = parse_time(raw_time)
cmd = parse_command(raw_command, available_commands)
# add a ScheduledCommand to the resulting list
tc = ScheduledCommand(delta, raw_command, cmd)
timed_commands.append(tc)
return timed_commands
def parse_excel_file(path, available_commands):
''' parses an time table in an excel file into a list of ScheduledCommands
path:
path to excel file
the time must be in the first column of the first sheet, the command in
the second column
available_commands:
a dictionary containing the available human readable commands as
keys and the commands to send as values
{'open': 0, 'close': 1}
returns a list consisting of ScheduledCommands
'''
workbook = load_workbook(path, read_only=True)
sheets = workbook.get_sheet_names()
sheet_name = sheets[0]
sheet = workbook[sheet_name]
is_header = True
timed_commands = []
for i, row in enumerate(sheet.rows):
delta = parse_time_cell(row[0])
print(i, row[0].value, delta)
if delta is None:
# no time in the cell
if is_header:
# this is still a header
continue
else:
# not the header and not a time cell
# premature end of list
break
else:
# a time in the cell, this is not a header any more
is_header = False
raw_command = row[1].value
cmd = parse_command(raw_command, available_commands)
# add a ScheduledCommand to the resulting list
tc = ScheduledCommand(delta, raw_command, cmd)
timed_commands.append(tc)
return timed_commands
def parse_time_cell(excel_cell):
if excel_cell.is_date:
return time_as_timedelta(excel_cell.value)
if isinstance(excel_cell.value, float):
return timedelta(days=excel_cell.value)
try:
delta = parse_time(excel_cell.value)
return delta
except ValueError:
return None
def parse_time(time_str):
''' parses a string to extract the time information
time_str:
string representaion of a time like '00:03.0'
returns a timedelta object, e.g. timedelta(seconds=3)
'''
# try the available time formats
for format_str in TIMETABLE_FORMATS:
try:
time_obj = datetime.strptime(time_str, format_str)
break
except ValueError:
pass
else:
msg = "time data '{}' does not match any format".format(time_str)
raise ValueError(msg)
return time_as_timedelta(time_obj)
def time_as_timedelta(time_object):
''' converts a time object to a timedelta '''
return timedelta(
hours=time_object.hour,
minutes=time_object.minute,
seconds=time_object.second,
microseconds=time_object.microsecond
)
def parse_command(command, available_commands):
''' parses a string and checks if it is a valid command
raw_command:
human readable representation of the command
available_commands:
a dictionary containing the available human readable commands as
keys and the commands to send as values
{'open': 0, 'close': 1}
returns the command to send over serial
'''
try:
cmd = available_commands[command.lower()]
except KeyError:
msg = "unknown command '{}'".format(command)
raise ValueError(msg)
if isinstance(cmd, str):
cmd = cmd.encode('utf-8')
return cmd
def find_arduino_port():
''' returns the port where an arduino is connected '''
# some regular expressions matching arduino and genuino
tmp = serial.tools.list_ports.grep('.+uino.+')
port_list = list(tmp)
# rais an error, if no or more than one arduinos are found
if len(port_list) == 0:
raise IOError('no arduino port found')
elif len(port_list) > 1:
raise IOError('{} arduino ports found'.format(len(port_list)))
# return only the device name
port_info = port_list[0]
return port_info.device
def send_command(scheduled_command, serial_connection):
''' sends a command over a serial connection '''
serial_connection.write(scheduled_command.cmd)
print(str(scheduled_command.delta), scheduled_command.command)
def run(scheduled_commands, serial_connection):
''' run scheduled commands
scheduled_commands:
list of ScheduledCommands
serial_connection:
serial connection object to send commands
'''
# use a new event loop every time
# if only one eventloop is used, the interactive use of run()
# does not work since the loop is closed in the end
loop = asyncio.new_event_loop()
for scheduled_command in scheduled_commands:
loop.call_later(
scheduled_command.delta.total_seconds(),
send_command,
scheduled_command,
serial_connection
)
# loop should stop one second after last command in list
timed_loop_stop = scheduled_command.delta + timedelta(seconds=1)
loop.call_later(timed_loop_stop.total_seconds(), loop.stop)
try:
loop.run_forever()
except KeyboardInterrupt:
loop.stop()
loop.close()
class TimedCommands(object):
''' lightweight encapsulation of the functions in the module '''
def __init__(self,
timetable,
available_commands,
port=None,
baudrate=9600,
**serial_kwargs):
''' parses a time table and establishes a serial connection
timetable:
a file path to an excel file or
a textual representaion of time and commands like
00:12.0 close
00:14.0 open
00:16.0 close
00:18.0 open
available_commands:
a dictionary containing the available human readable commands
as keys and the commands to send as values.
If an other iterable with strings is provided, the lowercase
version of the strings will be used as human readable commands
and the first letter of the strings as commands to send over the
wire.
port [None]
port where the Arduino is connected
if port is None, the ports are scanned for an Arduino
baudrate [9600]
speed of the serial connection
serial_kwargs
dictionary with further arguments for the serial connection
'''
# make sure the available commands are a dictionary
cmd_dict = self._ensure_command_dict(available_commands)
# parse the time table into something suitable
if '.xls' in timetable:
self.commands = parse_excel_file(timetable, cmd_dict)
else:
self.commands = parse_time_table(timetable, cmd_dict)
# establish the serial connection
port = port or find_arduino_port()
self.serial = serial.Serial(port, baudrate, **serial_kwargs)
def _ensure_command_dict(self, iterable):
''' ensures, that the available commands are a dictionary
iterable:
if the itarable is a dictionary, the keys will be
transformed to lowercase
if it is not a dictionary, the lowercase version of the
items is used as a human readable command and the first
character of this command will be sent over the wire
'''
try:
if isinstance(iterable, dict):
pairs = list(iterable.items())
keys = [key.lower() for key, value in pairs]
values = [value for key, value in pairs]
else:
tmp = (str(item) for item in iterable)
keys = [item.lower() for item in tmp]
values = [item[0] for item in keys]
pairs = zip(keys, values)
return dict(pairs)
except:
msg = 'available commands should be a list or dict of strings'
raise TypeError(msg)
def run(self):
''' run the scheduled commands '''
run(self.commands, self.serial)
class TrapControl(TimedCommands):
''' A simple, stripped down version for the magnetic plug trap '''
def __init__(self, timetable):
super().__init__(timetable, ['open', 'close'])

344
arduino_timetable/timetable.py
Unescape View File

@ -0,0 +1,344 @@
''' Controlling an Arduino with a time table '''
import asyncio
import serial
import serial.tools.list_ports
from collections import namedtuple
from datetime import datetime, timedelta
from openpyxl import load_workbook
# example time table for the TrapControl
example = '''
00:00.0 close
00:06.0 open
00:08.0 close
00:10.0 open
00:12.0 close
00:14.0 open
00:16.0 close
00:18.0 open
00:20.0 close
00:22.0 open
'''
# datetime formats for timetable parsing
TIMETABLE_FORMATS = ['%M:%S', '%M:%S.%f', '%H:%M:%S', '%H:%M:%S.%f']
# named tuple for a single scheduled command
ScheduledCommand = namedtuple('ScheduledCommand', 'delta command cmd')
def parse_time_table(timetable, available_commands=None):
''' parses a (text) time table into a list of ScheduledCommands
timetable:
a textual representaion of time and commands like
00:12.0 close
00:14.0 open
00:16.0 close
00:18.0 open
available_commands:
a dictionary containing the available human readable commands as
keys and the commands to send as values
{'open': 0, 'close': 1}
if None is provided, the check is skipped and the first lower case
letter used
returns a list consisting of ScheduledCommands
'''
# split the time table text by line break
raw_lines = timetable.split('\n')
# remove surrounding white space
content_lines = (line.strip() for line in raw_lines)
# remove empty lines
lines = (line for line in content_lines if line)
timed_commands =[]
for line in lines:
# split the lines into time and command
try:
raw_time, raw_command = line.split(None, 1)
except ValueError:
msg = "error in line '{}'".format(line)
raise ValueError(msg)
# parse time and commands
delta = parse_time(raw_time)
cmd = parse_command(raw_command, available_commands)
# add a ScheduledCommand to the resulting list
tc = ScheduledCommand(delta, raw_command, cmd)
timed_commands.append(tc)
return timed_commands
def parse_excel_file(path, available_commands=None):
''' parses an time table in an excel file into a list of ScheduledCommands
path:
path to excel file
the time must be in the first column of the first sheet, the command in
the second column
available_commands:
a dictionary containing the available human readable commands as
keys and the commands to send as values
{'open': 0, 'close': 1}
if None is provided, the check is skipped and the first lower case
letter used
returns a list consisting of ScheduledCommands
'''
workbook = load_workbook(path, read_only=True)
sheets = workbook.get_sheet_names()
sheet_name = sheets[0]
sheet = workbook[sheet_name]
is_header = True
timed_commands = []
for i, row in enumerate(sheet.rows):
delta = parse_time_cell(row[0])
if delta is None:
# no time in the cell
if is_header:
# this is still a header
continue
else:
# not the header and not a time cell
# premature end of list
break
else:
# a time in the cell, this is not a header any more
is_header = False
raw_command = row[1].value
cmd = parse_command(raw_command, available_commands)
# add a ScheduledCommand to the resulting list
tc = ScheduledCommand(delta, raw_command, cmd)
timed_commands.append(tc)
return timed_commands
def parse_time_cell(excel_cell):
''' parse an excel cell that contains a time value
if no time value is found in the excel cell None will be returned
'''
if excel_cell.is_date:
# the excel value is a datetime object
return time_as_timedelta(excel_cell.value)
if isinstance(excel_cell.value, float):
# sometimes excel stores times as a float value
return timedelta(days=excel_cell.value)
try:
# if it is not a common date fomat, try to parse it
delta = parse_time(excel_cell.value)
return delta
except ValueError:
return None
def parse_time(time_str):
''' parses a string to extract the time information
time_str:
string representaion of a time like '00:03.0'
returns a timedelta object, e.g. timedelta(seconds=3)
'''
# try the available time formats
time_str = str(time_str)
for format_str in TIMETABLE_FORMATS:
try:
time_obj = datetime.strptime(time_str, format_str)
break
except ValueError:
pass
else:
msg = "time data '{}' does not match any format".format(time_str)
raise ValueError(msg)
return time_as_timedelta(time_obj)
def time_as_timedelta(time_object):
''' converts a time object to a timedelta '''
return timedelta(
hours=time_object.hour,
minutes=time_object.minute,
seconds=time_object.second,
microseconds=time_object.microsecond
)
def parse_command(command, available_commands=None):
''' parses a string and checks if it is a valid command
raw_command:
human readable representation of the command
available_commands:
a dictionary containing the available human readable commands as
keys and the commands to send as values
{'open': 0, 'close': 1}
if None is provided, the check is skipped and the first lower case
letter used
returns the command to send over serial
'''
if available_commands is None:
cmd = str(command)[0]
else:
try:
cmd = available_commands[command.lower()]
except KeyError:
msg = "unknown command '{}'".format(command)
raise ValueError(msg)
if isinstance(cmd, str):
cmd = cmd.encode('utf-8')
return cmd
def find_arduino_port():
''' returns the port where an arduino is connected '''
# some regular expressions matching arduino and genuino
tmp = serial.tools.list_ports.grep('.+uino.+')
port_list = list(tmp)
# rais an error, if no or more than one arduinos are found
if len(port_list) == 0:
raise IOError('no arduino port found')
elif len(port_list) > 1:
raise IOError('{} arduino ports found'.format(len(port_list)))
# return only the device name
port_info = port_list[0]
return port_info.device
def send_command(scheduled_command, serial_connection):
''' sends a command over a serial connection '''
serial_connection.write(scheduled_command.cmd)
print(str(scheduled_command.delta), scheduled_command.command)
def run(scheduled_commands, serial_connection):
''' run scheduled commands
scheduled_commands:
list of ScheduledCommands
serial_connection:
serial connection object to send commands
'''
# use a new event loop every time
# if only one eventloop is used, the interactive use of run()
# does not work since the loop is closed in the end
loop = asyncio.new_event_loop()
for scheduled_command in scheduled_commands:
loop.call_later(
scheduled_command.delta.total_seconds(),
send_command,
scheduled_command,
serial_connection
)
# loop should stop one second after last command in list
timed_loop_stop = scheduled_command.delta + timedelta(seconds=1)
loop.call_later(timed_loop_stop.total_seconds(), loop.stop)
try:
loop.run_forever()
except KeyboardInterrupt:
loop.stop()
loop.close()
class TimedCommands(object):
''' lightweight encapsulation of the functions in the module '''
def __init__(self,
timetable,
available_commands,
port=None,
baudrate=9600,
**serial_kwargs):
''' parses a time table and establishes a serial connection
timetable:
a file path to an excel file or
a textual representaion of time and commands like
00:12.0 close
00:14.0 open
00:16.0 close
00:18.0 open
available_commands:
a dictionary containing the available human readable commands
as keys and the commands to send as values.
If an other iterable with strings is provided, the lowercase
version of the strings will be used as human readable commands
and the first letter of the strings as commands to send over the
wire.
port [None]
port where the Arduino is connected
if port is None, the ports are scanned for an Arduino
baudrate [9600]
speed of the serial connection
serial_kwargs
dictionary with further arguments for the serial connection
'''
# make sure the available commands are a dictionary
cmd_dict = self._ensure_command_dict(available_commands)
# parse the time table into something suitable
if '.xls' in timetable:
self.commands = parse_excel_file(timetable, cmd_dict)
else:
self.commands = parse_time_table(timetable, cmd_dict)
# establish the serial connection
port = port or find_arduino_port()
self.serial = serial.Serial(port, baudrate, **serial_kwargs)
def _ensure_command_dict(self, iterable):
''' ensures, that the available commands are a dictionary
iterable:
if the itarable is a dictionary, the keys will be
transformed to lowercase
if it is not a dictionary, the lowercase version of the
items is used as a human readable command and the first
character of this command will be sent over the wire
'''
try:
if isinstance(iterable, dict):
pairs = list(iterable.items())
keys = [key.lower() for key, value in pairs]
values = [value for key, value in pairs]
else:
tmp = (str(item) for item in iterable)
keys = [item.lower() for item in tmp]
values = [item[0] for item in keys]
pairs = zip(keys, values)
return dict(pairs)
except:
msg = 'available commands should be a list or dict of strings'
raise TypeError(msg)
def run(self):
''' run the scheduled commands '''
run(self.commands, self.serial)
def close(self):
''' closes an open serial connection '''
self.serial.close()

5
run_magnetic_trap.py
Unescape View File

@ -0,0 +1,5 @@
from arduino_timetable import run_application
if __name__ == '__main__':
arduino_commands = {'open': 'o', 'close': 'c'}
run_application(arduino_commands, "It's a trap!")
Write Preview
Loading…
Cancel
Save