# coding: utf-8 """ Implements an SQL database writer and reader for storing CAN messages. .. note:: The database schema is given in the documentation of the loggers. """ from __future__ import absolute_import import sys import time import threading import logging import sqlite3 from can.listener import BufferedReader from can.message import Message from .generic import BaseIOHandler log = logging.getLogger('can.io.sqlite') if sys.version_info.major < 3: # legacy fallback for Python 2 memoryview = buffer class SqliteReader(BaseIOHandler): """ 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:`~builtin.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, table_name="messages"): """ :param file: a `str` or since Python 3.7 a 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(SqliteReader, self).__init__(file=None) self._conn = sqlite3.connect(file) self._cursor = self._conn.cursor() self.table_name = table_name def __iter__(self): for frame_data in self._cursor.execute("SELECT * FROM {}".format(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), 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("SELECT COUNT(*) FROM {}".format(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("SELECT * FROM {}".format(self.table_name)).fetchall() return (SqliteReader._assemble_message(frame) for frame in result) def stop(self): """Closes the connection to the database. """ super(SqliteReader, self).stop() self._conn.close() class SqliteWriter(BaseIOHandler, 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 actally writtem 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, table_name="messages"): """ :param file: a `str` or since Python 3.7 a 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. """ super(SqliteWriter, self).__init__(file=None) self.table_name = table_name self._db_filename = file self._stop_running_event = threading.Event() self._writer_thread = threading.Thread(target=self._db_writer_thread) self._writer_thread.start() self.num_frames = 0 self.last_write = time.time() 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(""" CREATE TABLE IF NOT EXISTS {} ( ts REAL, arbitration_id INTEGER, extended INTEGER, remote INTEGER, error INTEGER, dlc INTEGER, data BLOB ) """.format(self.table_name)) self._conn.commit() self._insert_template = "INSERT INTO {} VALUES (?, ?, ?, ?, ?, ?, ?)".format(self.table_name) 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 else: # 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 an block. """ BufferedReader.stop(self) self._stop_running_event.set() self._writer_thread.join() BaseIOHandler.stop(self)