biometrics_tracker.model.persistence module

Index

class biometrics_tracker.model.persistence.DataBase(filename: str, queues: Queues, block_req_queue: bool)

Bases: Thread

Establishes a connection to the sqlite3 database and provides methods that insert/update the variable tables. The database contains the following tables: - PEOPLE - one row for each person whose biometrics are tracked by the application - TRACK_CFG - one row for each metric tracked for each person. This table has a foreign key relationship with the PEOPLE table. - DATAPOINTS - one row for each instance of a recorded metric for each person. This table has a foreign key relationship with the PEOPLE table - SCHEDULE - one row for each schedule entry for the scheduled collection of biometrics data. This table has a foreign key relationship with the PEOPLE table

static build_schedule(person_id: str, row: Row) → ScheduleEntry

Creates a ScheduleEntry instance from a row retrieved from the SCHEDULE table

Parameters

person_id (str) – the ID of the Person whose Schedule entries were retrieved
row (sql.Row) – the row retrieved from the database

Returns

the ScheduleEntry instance

Return type

biometrics_tracker.model.datapoints.ScheduleEntry

close()

Close the database connection

Returns: None

close_db() → None

Close the database connection and set the closed boolean to True, when will cause the run method to exit

Returns: None

commit()

Executes a COMMIT statement

Returns: None

create_db()

Executes a series of CREATE TABLE and CREATE INDEX statements to create the database

Returns: None

delete_schedule_for_person_seq(person_id: str, seq_nbr: int)

Delete a single schedule entry

Parameters

person_id (str) – ID of the person whose schedule entry will be deleted
seq_nbr (int) – sequence number of the schedule entry will be deleted

Returns

None

delete_schedules_for_person(person_id: str)

Delete all schedule entries for the specified person

Parameters: person_id (str) – ID of the person whose schedule entries will be deleted
Returns: None

delete_tracking_cfg_for_person(person_id)

Deletes all the rows from the TRACK_CFG table that are related to the specified person

Parameters: person_id (str) – the id of the person whose entries will be deleted
Returns: None

drop_db()

Executes a series of DROP TABLE and DROP INDEX statements to un-delete the entire database

Returns: None

get_connection(filename: str)

Establishes a connection to the sqlite3 database

Parameters: filename (str) – the filename of the file containing the database
Returns: an sql.connection object
Return type: sqlite3.Connection

handle_dp_req(msg: Union[DataPointReqMsg, DataPointMsg]) → Optional[DataPointRespMsg]

Handle DataPoint requests - deletes, inserts and updates

Parameters: msg (Union[biometrics_tracker.ipc.messages.DataPointReqMsg, biometrics_tracker.ipc.messages.DataPointMsg]) – a message containing the request info
Returns: a message containing the response info
Return type: Optional[biometrics_tracker.ipc.messages.DataPointRespMsg]

handle_person_req(msg: PersonReqMsg) → Optional[Union[PersonMsg, PeopleMsg]]

Handle PersonReqMsg tasks - retrieve a list of Person instances or a single instance

Parameters: msg (biometrics_tracker.ipc.messages.PersonReqMsg) – the message containing the request info
Returns: if appropriate, returns a response message containing the requested data
Return type: Optional[Union[biometrics_tracker.ipc.messages.PersonMsg,biometrics_tracker.ipc.messages.PeopleMsg]]:

handle_schedule_req(msg: Union[ScheduleEntryReqMsg, ScheduleEntryMsg]) → ScheduleEntriesMsg: Handle ScheduleEntry requests - deletes retrievals and upates :param msg: :return: a message containing the response :rtype: biometrics_tracker.ipc.messages.ScheduleEntriesMsg

handle_track_cfg_req(msg: Union[TrackingConfigReqMsg, TrackingConfigMsg]) → None

Handle TrackConfig requests - deletes and updates

Parameters: msg – a message containing the request info

:type msg:Union[biometrics_tracker.ipc.messages.TrackingConfigReqMsg, biometrics_tracker.ipc.messages.TrackingConfigMsg] :return: None

insert_datapoint(datapoint: DataPoint)

Inserts the data from an instance of a subclass of model.datapoints.DataPoint to the DATAPOINTS table

Parameters: datapoint (biometrics_tracker.model.datapoints.DataPoint) – datapoint info to be inserted
Returns: None

insert_person(person: Person)

Inserts the info in the provided model.datapoints.Person instance to the PEOPLE table

Parameters: person (biometrics_tracker.model.datapoints.Person) – info for person to be inserted
Returns: None

insert_schedule_entry(sched_entry: ScheduleEntry)

Inserts the info in the provided biometrics_tracking.model.datapoints.ScheduleEntry instance to the SCHEDULE table.

Parameters: sched_entry (biometrics_tracking.model.datapoints.ScheduleEntry) – info for the schedule entry to be inserted/updated
Returns: None

static retrieve_datapoints(rows: list) → list[biometrics_tracker.model.datapoints.DataPoint]

Creates a list of DataPoints from rows retrieved from the DATAPOINTS table

Parameters: rows (list[list[tuple[]]]) – the rows retrieved by a SELECT statement
Returns: a list of DataPoints
Return type: list[biometrics_tracker.model.datapoints.DataPoint]

retrieve_datapoints_for_person(person_id: str, start_date: datetime, end_date: datetime) → list[biometrics_tracker.model.datapoints.DataPoint]

Retrieves rows related to the specified person id from the DATAPOINTS table, returning a list of instances of subclasses of model.datapoints.DataPoint

Parameters

person_id (str) – indicates the person whose data is to be retrieved
start_date (datetime.datetime) – datetime of the first row to be retrieved
end_date (datetime.datetime) – datetime of the last row to be retrieved

Returns

a list of instances of subclasses of model.datapoints.DataPoint

Return type

list[model.datapoints.UOM]

retrieve_datapoints_where(person_id: Optional[str] = None, start_datetime: Optional[datetime] = None, end_datetime: Optional[datetime] = None, dp_type: Optional[DataPointType] = None) → list[biometrics_tracker.model.datapoints.DataPoint]

This method was created to provide more flexibility in the selection criteria for DataPoints. Selection for Person, date range and DataPointType can be mixed and matches. A parameter value of None indicates that the column will not be included in the SELECT statement where clause.

Parameters

person_id (Optional[str]) – the optional ID of the person to select rows for
start_datetime (Optional[datetime]) – the optional earliest date/time to select rows for
end_datetime (Optional[datetime]) – the optional most recent date/time to select rows for
dp_type (Optional[biometrics_tracker.model.datapoints.DataPointType]) – the optional DataPointType to select rows for

Returns

a list of DataPoints that meet the provided selection criterial

Return type

list[biometrics_tracker.model.datapoints.DataPoint]

retrieve_people() → list[tuple[str, str]]

Retrieves a list of (id, name) tuples from the PEOPLE table. This list can be used to implement a selection list for the people whose biometrics are tracked by the application

Returns: a list of (string, string) tuples, the first element being the person id, the second the person name
Return type: list[tuple[str, str]]

retrieve_person(person_id: str)

Retrieve a row from the PEOPLE table for the specified person id

Parameters: person_id (str) – The id of the row to be retrieved
Returns: retrieved Person info
Return type: biometrics_tracker.model.datapoints.Person

retrieve_schedule_for_person_seq(person_id: str, seq_nbr: int) → Optional[ScheduleEntry]

Retrieves a single ScheduleEntry matching the supplied Person ID and SecheduleEntry sequence number

Parameters

person_id (str) – ID of the person whose ScheduleEntry will be retrieved
seq_nbr (int) – sequence number of the ScheduleEntry to be retrieved

Returns

the retrieved ScheduleEntry or None

Return type

Optional[biometrics_tracker.model.datapoints.ScheduleEntry]

retrieve_schedules_for_person(person_id) → list[biometrics_tracker.model.datapoints.ScheduleEntry]

Retrieve the schedule entries associated with the specified person

Parameters: person_id (str) – ID of the person whose schedule entries will be retrieved
Returns: biometrics_tracker.ipc.messages.ScheduleEntriesMsg

run()

This method is invoked when the DataBase thread is started

Returns: None

update_datapoint(datapoint: DataPoint, key_timestamp: datetime)

Updates a DATAPOINTS table row with the data from an instance of a subclass of model.datapoints.DataPoint

Parameters: datapoint (an instance of one of the classes in model.datapoints.dptypes_union) – an datapoint to be upated
Returns: None

update_person(person: Person)

Updates a PERSON table row with the provided model.datapoints.Person instance

Parameters: person (biometrics_tracker.model.datapoints.Person) – info for person to be updated
Returns: None

update_schedule_entry(sched_entry: ScheduleEntry)

Inserts the info in the provided biometrics_tracking.model.datapoints.ScheduleEntry instance to the SCHEDULE table.

Parameters: sched_entry (biometrics_tracking.model.datapoints.ScheduleEntry) – info for the schedule entry to be inserted/updated
Returns: None

update_schedule_last_triggered(person_id: str, seq_nbr: int, last_triggered: datetime)

Update the last triggered date in the specified Schedule Entry

Parameters

person_id (str) – ID of the person related to the Schedule Entry
seq_nbr (int) – Sequence Number related to the Schedule Entry
last_triggered (datetime) – the datetime the schedule entry was triggered

Returns

None

upsert_tracking_cfg(person_id: str, tracking_cfg: TrackingConfig)

Inserts a row into the TRACK_CFG table

Parameters

person_id (str) – the id of the person related to the Tracking Config info
tracking_cfg (biometrics_tracker.model.datapoints.TrackingConfig) – Tracking configuration to be inserted

Returns

None