forked from hardbyte/python-can
-
Notifications
You must be signed in to change notification settings - Fork 6
Expand file tree
/
Copy pathCAN.py
More file actions
134 lines (101 loc) · 3.79 KB
/
Copy pathCAN.py
File metadata and controls
134 lines (101 loc) · 3.79 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
"""
The core of python-can - contains implementations of all
the major classes in the library, which form abstractions of the
functionality provided by each CAN interface.
Copyright (C) 2010 Dynamic Controls
"""
from __future__ import print_function
import logging
try:
import queue
except ImportError:
import Queue as queue
log = logging.getLogger('can')
log.debug("Loading python-can")
def set_logging_level(level_name=None):
"""Set the logging level for python-can.
Expects one of: 'critical', 'error', 'warning', 'info', 'debug', 'subdebug'
"""
try:
log.setLevel(getattr(logging, level_name.upper()))
except AttributeError:
log.setLevel(logging.DEBUG)
log.debug("Logging set to {}".format(level_name))
logging.basicConfig()
class Listener(object):
def on_message_received(self, msg):
raise NotImplementedError(
"{} has not implemented on_message_received".format(
self.__class__.__name__)
)
def __call__(self, msg):
return self.on_message_received(msg)
class BufferedReader(Listener):
"""
A BufferedReader is a subclass of :class:`~can.Listener` which implements a
**message buffer**: that is, when the :class:`can.BufferedReader` instance is
notified of a new message it pushes it into a queue of messages waiting to
be serviced.
"""
def __init__(self):
self.buffer = queue.Queue(0)
def on_message_received(self, msg):
self.buffer.put(msg)
def get_message(self, timeout=0.5):
"""
Attempts to retrieve the latest message received by the instance. If no message is
available it blocks for 0.5 seconds or until a message is received (whichever
is shorter), and returns the message if there is one, or None if there is not.
"""
try:
return self.buffer.get(block=True, timeout=timeout)
except queue.Empty:
return None
class Printer(Listener):
"""
The Printer class is a subclass of :class:`~can.Listener` which simply prints
any messages it receives to the terminal.
:param output_file: An optional file to "print" to.
"""
def __init__(self, output_file=None):
if output_file is not None:
log.info("Creating log file '{}' ".format(output_file))
output_file = open(output_file, 'wt')
self.output_file = output_file
def on_message_received(self, msg):
if self.output_file is not None:
self.output_file.write(str(msg) + "\n")
else:
print(msg)
def __del__(self):
self.output_file.write("\n")
if self.output_file:
self.output_file.close()
class CSVWriter(Listener):
"""Writes a comma separated text file of
timestamp, arbitrationid, flags, dlc, data
for each messages received.
"""
def __init__(self, filename):
self.csv_file = open(filename, 'wt')
# Write a header row
self.csv_file.write("timestamp, arbitrationid, flags, dlc, data")
def on_message_received(self, msg):
row = ','.join([msg.timestamp,
msg.arbitration_id,
msg.flags,
msg.dlc,
msg.data])
self.csv_file.write(row + '\n')
def __del__(self):
self.csv_file.close()
super(CSVWriter, self).__del__()
class SqliteWriter(Listener):
"""TODO"""
def __init__(self, filename):
self.db_file = open(filename, 'wt')
# create table structure
raise NotImplementedError("TODO")
def on_message_received(self, msg):
# add row
raise NotImplementedError("TODO")