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
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
|
"""
Implements an SQL database writer and reader for storing CAN messages.
.. note:: The database schema is given in the documentation of the loggers.
"""
import logging
import sqlite3
import threading
import time
from typing import Any, Generator
from can.listener import BufferedReader
from can.message import Message
from ..typechecking import StringPathLike
from .generic import MessageReader, MessageWriter
log = logging.getLogger("can.io.sqlite")
class SqliteReader(MessageReader):
"""
Reads recorded CAN messages from a simple SQL database.
This class can be iterated over or used to fetch all messages in the
database with :meth:`~SqliteReader.read_all`.
Calling :func:`len` on this object might not run in constant time.
:attr str table_name: the name of the database table used for storing the messages
.. note:: The database schema is given in the documentation of the loggers.
"""
def __init__(
self,
file: StringPathLike,
table_name: str = "messages",
**kwargs: Any,
) -> None:
"""
:param file: a `str` path like object that points
to the database file to use
:param str table_name: the name of the table to look for the messages
.. warning:: In contrary to all other readers/writers the Sqlite handlers
do not accept file-like objects as the `file` parameter.
It also runs in ``append=True`` mode all the time.
"""
super().__init__(file=None)
self._conn = sqlite3.connect(file)
self._cursor = self._conn.cursor()
self.table_name = table_name
def __iter__(self) -> Generator[Message, None, None]:
for frame_data in self._cursor.execute(f"SELECT * FROM {self.table_name}"):
yield SqliteReader._assemble_message(frame_data)
@staticmethod
def _assemble_message(frame_data):
timestamp, can_id, is_extended, is_remote, is_error, dlc, data = frame_data
return Message(
timestamp=timestamp,
is_remote_frame=bool(is_remote),
is_extended_id=bool(is_extended),
is_error_frame=bool(is_error),
arbitration_id=can_id,
dlc=dlc,
data=data,
)
def __len__(self):
# this might not run in constant time
result = self._cursor.execute(f"SELECT COUNT(*) FROM {self.table_name}")
return int(result.fetchone()[0])
def read_all(self):
"""Fetches all messages in the database.
:rtype: Generator[can.Message]
"""
result = self._cursor.execute(f"SELECT * FROM {self.table_name}").fetchall()
return (SqliteReader._assemble_message(frame) for frame in result)
def stop(self):
"""Closes the connection to the database."""
super().stop()
self._conn.close()
class SqliteWriter(MessageWriter, BufferedReader):
"""Logs received CAN data to a simple SQL database.
The sqlite database may already exist, otherwise it will
be created when the first message arrives.
Messages are internally buffered and written to the SQL file in a background
thread. Ensures that all messages that are added before calling :meth:`~can.SqliteWriter.stop()`
are actually written to the database after that call returns. Thus, calling
:meth:`~can.SqliteWriter.stop()` may take a while.
:attr str table_name: the name of the database table used for storing the messages
:attr int num_frames: the number of frames actually written to the database, this
excludes messages that are still buffered
:attr float last_write: the last time a message war actually written to the database,
as given by ``time.time()``
.. note::
When the listener's :meth:`~SqliteWriter.stop` method is called the
thread writing to the database will continue to receive and internally
buffer messages if they continue to arrive before the
:attr:`~SqliteWriter.GET_MESSAGE_TIMEOUT`.
If the :attr:`~SqliteWriter.GET_MESSAGE_TIMEOUT` expires before a message
is received, the internal buffer is written out to the database file.
However if the bus is still saturated with messages, the Listener
will continue receiving until the :attr:`~can.SqliteWriter.MAX_TIME_BETWEEN_WRITES`
timeout is reached or more than
:attr:`~can.SqliteWriter.MAX_BUFFER_SIZE_BEFORE_WRITES` messages are buffered.
.. note:: The database schema is given in the documentation of the loggers.
"""
GET_MESSAGE_TIMEOUT = 0.25
"""Number of seconds to wait for messages from internal queue"""
MAX_TIME_BETWEEN_WRITES = 5.0
"""Maximum number of seconds to wait between writes to the database"""
MAX_BUFFER_SIZE_BEFORE_WRITES = 500
"""Maximum number of messages to buffer before writing to the database"""
def __init__(
self,
file: StringPathLike,
table_name: str = "messages",
**kwargs: Any,
) -> None:
"""
:param file: a `str` or path like object that points
to the database file to use
:param str table_name: the name of the table to store messages in
.. warning:: In contrary to all other readers/writers the Sqlite handlers
do not accept file-like objects as the `file` parameter.
"""
if kwargs.get("append", False):
raise ValueError(
f"The append argument should not be used in "
f"conjunction with the {self.__class__.__name__}."
)
super().__init__(file=None)
self.table_name = table_name
self._db_filename = file
self._stop_running_event = threading.Event()
self._conn = None
self._writer_thread = threading.Thread(target=self._db_writer_thread)
self._writer_thread.start()
self.num_frames = 0
self.last_write = time.time()
self._insert_template = (
f"INSERT INTO {self.table_name} VALUES (?, ?, ?, ?, ?, ?, ?)"
)
def _create_db(self):
"""Creates a new databae or opens a connection to an existing one.
.. note::
You can't share sqlite3 connections between threads (by default)
hence we setup the db here. It has the upside of running async.
"""
log.debug("Creating sqlite database")
self._conn = sqlite3.connect(self._db_filename)
# create table structure
self._conn.cursor().execute(
f"""CREATE TABLE IF NOT EXISTS {self.table_name}
(
ts REAL,
arbitration_id INTEGER,
extended INTEGER,
remote INTEGER,
error INTEGER,
dlc INTEGER,
data BLOB
)"""
)
self._conn.commit()
def _db_writer_thread(self):
self._create_db()
try:
while True:
messages = [] # reset buffer
msg = self.get_message(self.GET_MESSAGE_TIMEOUT)
while msg is not None:
# log.debug("SqliteWriter: buffering message")
messages.append(
(
msg.timestamp,
msg.arbitration_id,
msg.is_extended_id,
msg.is_remote_frame,
msg.is_error_frame,
msg.dlc,
memoryview(msg.data),
)
)
if (
time.time() - self.last_write > self.MAX_TIME_BETWEEN_WRITES
or len(messages) > self.MAX_BUFFER_SIZE_BEFORE_WRITES
):
break
# just go on
msg = self.get_message(self.GET_MESSAGE_TIMEOUT)
count = len(messages)
if count > 0:
with self._conn:
# log.debug("Writing %d frames to db", count)
self._conn.executemany(self._insert_template, messages)
self._conn.commit() # make the changes visible to the entire database
self.num_frames += count
self.last_write = time.time()
# check if we are still supposed to run and go back up if yes
if self._stop_running_event.is_set():
break
finally:
self._conn.close()
log.info("Stopped sqlite writer after writing %d messages", self.num_frames)
def stop(self):
"""Stops the reader an writes all remaining messages to the database. Thus, this
might take a while and block.
"""
BufferedReader.stop(self)
self._stop_running_event.set()
self._writer_thread.join()
MessageReader.stop(self)
|
