diff --git a/mosaiq/__init__.py b/mosaiq/__init__.py index d61a6c1..2e13334 100644 --- a/mosaiq/__init__.py +++ b/mosaiq/__init__.py @@ -28,4 +28,7 @@ from .scheduled_field import ScheduledField from .session import Session from .site_setup import SiteSetup -from .task import Task \ No newline at end of file +from .task import Task +from .episode import Episode +from .payer import Payer +from .pay_pay import PatPay \ No newline at end of file diff --git a/mosaiq/activity.py b/mosaiq/activity.py index abf41fc..6fdb4c5 100644 --- a/mosaiq/activity.py +++ b/mosaiq/activity.py @@ -13,7 +13,7 @@ #from tkinter import messagebox from .database import Database - +from pprint import pprint class Activity: # Returns a single activity matching the given database id (PRS_ID) (or None if no match). @@ -30,7 +30,7 @@ def find(cls, id): def find_by_code(cls, hsp_code): instance = None if len(hsp_code) > 0: - row = Database.fetch_one("SELECT * FROM CPT WHERE Hsp_Code = '{}'".format(str(hsp_Code))) + row = Database.fetch_one("SELECT * FROM CPT WHERE Hsp_Code = '{}'".format(str(hsp_code))) if row != None: instance = cls(row) return instance @@ -41,12 +41,21 @@ def __init__(self, row): self.prs_id = row['PRS_ID'] self.inactive = row['Status_Inactive'] self.code_group = row['CGroup'].rstrip() # If this crashes sometimes, we have to test if the string exists. + ''' self.code1 = row['Hsp_Code'] self.code2 = row['Hsp_Code1'] self.code3 = row['Hsp_Code2'] self.code4 = row['Hsp_Code3'] self.code5 = row['Hsp_Code4'] self.code6 = row['Hsp_Code5'] + ''' + # GC version + self.code1 = row['Hsp_Code'] + self.code2 = row['Hsp_Code2'] + self.code3 = row['Hsp_Code3'] + self.code4 = row['Hsp_Code4'] + self.code5 = row['Hsp_Code5'] + self.charge_code = row['CPT_Code'].rstrip() # If this crashes sometimes, we have to test if the string exists. self.abbreviation = row['Tiny_Desc'].rstrip() # If this crashes sometimes, we have to test if the string exists. self.title = row['Short_Desc'] diff --git a/mosaiq/appointment.py b/mosaiq/appointment.py index 0c470b8..8136e23 100644 --- a/mosaiq/appointment.py +++ b/mosaiq/appointment.py @@ -13,6 +13,10 @@ #from tkinter import messagebox from .database import Database +from .location import Location +from .task import Task +from .patient import Patient +#from pprint import pprint class Appointment: @@ -25,6 +29,30 @@ def find(cls, id): instance = cls(row) return instance + # Gives all appointment belonging to a specific Location_id (str) + # Excludes deleted appointments (suppressed = true). + # Excludes historic appointments (version != 0). + # Returns appointments sorted by their start_date parameter. + @classmethod + def for_location_id(cls, location_id): + appointments = list() + rows = Database.fetch_all("SELECT * FROM Schedule WHERE Location = '{}' AND Suppressed != {} AND Version = 0".format(location_id, 1)) # GCUK MSQ this is bit, not bool + for row in rows: + appointments.append(cls(row)) + return appointments + + # Gives all appointment belonging to a specific Location object + # Excludes deleted appointments (suppressed = true). + # Excludes historic appointments (version != 0). + # Returns appointments sorted by their start_date parameter. + @classmethod + def for_location(cls, location): + appointments = list() + rows = Database.fetch_all("SELECT * FROM Schedule WHERE Location = '{}' AND Suppressed != {} AND Version = 0".format(location.id, 1)) # GCUK MSQ this is bit, not bool + for row in rows: + appointments.append(cls(row)) + return appointments + # Gives all appointments belonging to the given patient. # Excludes deleted appointments (suppressed = true). # Excludes historic appointments (version != 0). @@ -32,7 +60,8 @@ def find(cls, id): @classmethod def for_patient(cls, patient): appointments = list() - rows = Database.fetch_all("SELECT * FROM Schedule WHERE Pat_ID1 = '{}' AND Suppressed != {} AND Version = 0".format(patient.id, True)) + #rows = Database.fetch_all("SELECT * FROM Schedule WHERE Pat_ID1 = '{}' AND Suppressed != {} AND Version = 0".format(patient.id, True)) + rows = Database.fetch_all("SELECT * FROM Schedule WHERE Pat_ID1 = '{}' AND Suppressed != {} AND Version = 0".format(patient.id, 1)) # GCUK MSQ this is bit, not bool for row in rows: appointments.append(cls(row)) return appointments @@ -43,6 +72,7 @@ def __init__(self, row): self.sch_id = row['Sch_Id'] self.sch_set_id = row['Sch_Set_Id'] self.related_appointment_id = row['Sch_Set_Id'] + self.activity_code_long = row['Activity'] self.activity_code = row['Activity'].rstrip() # If this crashes sometimes, we have to test if the string exists. self.start_date = row['App_DtTm'] self.location_id = row['Location'] @@ -134,6 +164,7 @@ def start(self): # Gives the task item referenced by this appointment. def task(self): if not self.instance_task: - self.instance_task = Task.find(self.task_id) + #self.instance_task = Task.find(self.task_id) + self.instance_task = Task.find(self.id) return self.instance_task \ No newline at end of file diff --git a/mosaiq/checklist.py b/mosaiq/checklist.py index 08c58bb..b740f84 100644 --- a/mosaiq/checklist.py +++ b/mosaiq/checklist.py @@ -13,8 +13,26 @@ #from tkinter import messagebox from .database import Database +from .patient import Patient +from .task import Task class Checklist: + + + def __repr__(self): + status = "complete" if self.complete else "incomplete" + suppressed = "suppressed" if self.suppressed else "active" + + # Dates are often datetime/None in Mosaiq extracts; keep readable and safe. + due = self.due_date.date() if self.due_date else None + done = self.completed_date.date() if self.completed_date else None + + return ( + f"Checklist(patient_id={self.patient().last_name!r}, " + f"task_id={self.task().description!r}, " + f"{status} " + f"due_date={due!r}, completed_date={done!r})" + ) # Returns a single checklist matching the given database id (Chk_id) (or None if no match). @classmethod @@ -24,14 +42,43 @@ def find(cls, id): if row != None: instance = cls(row) return instance + + # Gives all checklists with a given task_id across all patients + @classmethod + def of_type(cls, task_id, complete=None): + query = "SELECT * FROM Chklist WHERE TSK_ID = '{}'".format(task_id) + if complete is not None: + query += " AND Complete = '{}'".format(str(complete)) + checklists = list() + rows = Database.fetch_all(query) + for row in rows: + checklists.append(cls(row)) + return checklists # Gives all checklists belonging to the given patient. # Optionally the query may be restricted to a given task type (specified by id). @classmethod - def for_patient(cls, patient, task_id=None): + def for_patient(cls, patient, task_id=None, complete=None): query = "SELECT * FROM Chklist WHERE Pat_ID1 = '{}'".format(patient.id) if task_id: query += " AND TSK_ID = '{}'".format(task_id) + if complete is not None: + query += " AND Complete = '{}'".format(str(complete)) + checklists = list() + rows = Database.fetch_all(query) + for row in rows: + checklists.append(cls(row)) + return checklists + + # Gives all checklists for a staff/location/group + # Optionally the query may be restricted to a given task type (specified by id). + @classmethod + def for_responsible(cls, responsible, task_id=None, complete=None): + query = "SELECT * FROM Chklist WHERE Rsp_Staff_ID = '{}'".format(responsible.id) + if task_id: + query += " AND TSK_ID = '{}'".format(task_id) + if complete is not None: + query += " AND Complete = '{}'".format(str(complete)) checklists = list() rows = Database.fetch_all(query) for row in rows: diff --git a/mosaiq/database.py b/mosaiq/database.py index e5f4d9f..3690ce2 100644 --- a/mosaiq/database.py +++ b/mosaiq/database.py @@ -13,11 +13,14 @@ #from tkinter import messagebox import pymssql +import pyodbc +from pathlib import Path class Database: - # The Mosaiq SQL server address: - server = open(r'C:\temp\raystation-scripts\mosaiq\database.txt', "r").read() + # The Mosaiq SQL server address and (optional) database: + server = open(r'C:\temp\raystation-scripts\mosaiq\server.txt', "r").read() + database = open(r'C:\temp\raystation-scripts\mosaiq\database.txt').read() # The username to be used for access to the Mosaiq database: user = open(r'C:\temp\raystation-scripts\mosaiq\user.txt', "r").read() # The password to be used for access to the Mosaiq database: @@ -26,7 +29,7 @@ class Database: # Returns all rows matching the given query text (or an empty list if no match). @staticmethod def fetch_all(text): - conn = pymssql.connect(server=Database.server, user=Database.user, password=Database.password) + conn = pymssql.connect(server=Database.server, database=Database.database, user=Database.user, password=Database.password) cursor = conn.cursor(as_dict=True) cursor.execute(text) rows = list() @@ -39,7 +42,7 @@ def fetch_all(text): # Returns a single row matching the given query text (or None if no match). @staticmethod def fetch_one(text): - conn = pymssql.connect(server=Database.server, user=Database.user, password=Database.password) + conn = pymssql.connect(server=Database.server, database=Database.database, user=Database.user, password=Database.password) cursor = conn.cursor(as_dict=True) cursor.execute(text) row = cursor.fetchone() diff --git a/mosaiq/delivered_dose.py b/mosaiq/delivered_dose.py index c033abe..f22f6ee 100644 --- a/mosaiq/delivered_dose.py +++ b/mosaiq/delivered_dose.py @@ -47,7 +47,7 @@ def for_patient(cls, patient): @classmethod def for_prescription(cls, prescription): delivered_doses = list() - rows = Database.fetch_all("SELECT * FROM Dose_Hst WHERE SIT_ID = '{}'".format(field.id)) + rows = Database.fetch_all("SELECT * FROM Dose_Hst WHERE SIT_ID = '{}'".format(prescription.id)) for row in rows: delivered_doses.append(cls(row)) return delivered_doses @@ -56,7 +56,7 @@ def for_prescription(cls, prescription): @classmethod def for_scheduled_field(cls, scheduled_field): delivered_doses = list() - rows = Database.fetch_all("SELECT * FROM Dose_Hst WHERE PTC_ID = '{}'".format(field.id)) + rows = Database.fetch_all("SELECT * FROM Dose_Hst WHERE PTC_ID = '{}'".format(scheduled_field.id)) for row in rows: delivered_doses.append(cls(row)) return delivered_doses diff --git a/mosaiq/document.py b/mosaiq/document.py index 594ed4f..b1e963c 100644 --- a/mosaiq/document.py +++ b/mosaiq/document.py @@ -27,9 +27,12 @@ def find(cls, id): # Gives all documents belonging to the given patient. @classmethod - def for_patient(cls, patient): + def for_patient(cls, patient, type_id=None): + query = "SELECT * FROM Object WHERE Pat_ID1 = '{}'".format(patient.id) + if type_id: + query += " AND DocType = '{}'".format(type_id) documents = list() - rows = Database.fetch_all("SELECT * FROM Object WHERE Pat_ID1 = '{}'".format(patient.id)) + rows = Database.fetch_all(query) for row in rows: documents.append(cls(row)) return documents @@ -53,6 +56,7 @@ def __init__(self, row): self.document_id = row['OBJ_SET_ID'] self.version = row['Version'] self.institution_id = row['Inst_ID'] + self.document_template = row['DocumentTemplate'] # Convenience attributes: self.id = self.obj_id # Cache attributes: @@ -63,6 +67,7 @@ def __init__(self, row): self.instance_note = None self.instance_nr_pages = None self.instance_patient = None + self.document_type = None # The staff who approved the document. def approved_by(self): @@ -99,7 +104,7 @@ def file_format(self): # The file name of the document. def file_name(self): if not self.instance_file_name: - row = Database.fetch_one("SELECT * FROM ObjFilenames WHERE OBJ_ID = '{}'".format(str(id))) + row = Database.fetch_one("SELECT * FROM ObjFilenames WHERE OBJ_ID = '{}'".format(str(self.id))) if row != None: self.instance_file_name = row['eSCANFilename'] return self.instance_file_name @@ -118,6 +123,15 @@ def nr_pages(self): self.instance_nr_pages = row['PageNumber'] return self.instance_nr_pages + # The document type + def document_type(self): + if not self.document_type: + row = Database.fetch_one("SELECT * FROM ObjPhd WHERE OBJ_ID = '{}'".format(str(id))) + print(row) + if row != None: + self.document_type = row["Description"] + return self.document_type + # Gives the patient which this document belongs to. def patient(self): if not self.instance_patient: diff --git a/mosaiq/episode.py b/mosaiq/episode.py new file mode 100644 index 0000000..ebb1f65 --- /dev/null +++ b/mosaiq/episode.py @@ -0,0 +1,65 @@ +# encoding: utf8 + +# A class for reading Episode of Care data from the Mosaiq database. +# +# Authors: +# Ben George +# +# Python 3.6 + +from .database import Database + +#from pprint import pprint + +class Episode: + + # Returns a single task matching the given database id (TSK_ID) (or None if no match). + @classmethod + def find(cls, id): + instance = None + row = Database.fetch_one("SELECT * FROM Episode WHERE Epi_Id = '{}'".format(str(id))) + if row != None: + instance = cls(row) + return instance + + @classmethod + def for_patient(cls, patient): + instance = None + query = "SELECT * FROM Episode WHERE Pat_ID1 = '{}'".format(patient.id) + episodes = list() + rows = Database.fetch_all(query) + for row in rows: + episodes.append(cls(row)) + return episodes + + # Creates a Task instance from a task database row. + def __init__(self, row): + # Database attributes: + self.epi_id = row['Epi_Id'] + self.created_date = row['Create_DtTm'] + self.created_by_id = row['Create_ID'] + self.patient_id = row['Pat_ID1'] + self.edited_date = row['Edit_DtTm'] + self.edited_by_id = row['Edit_Id'] + self.comment = row['Comment']#.rstrip() # If this crashes sometimes, we have to test if the string exists. + self.first_treatment_date = row['FirstTx_DtTm'] + self.active_date = row['Active_DtTm'] + self.inactive_date = row['Inactive_DtTm'] + # Convenience attributes: + self.id = self.epi_id + # Cache attributes: + self.instance_created_by = None + self.instance_edited_by = None + + # The staff who created the task. + def created_by(self): + if not self.instance_created_by: + self.instance_created_by = Location.find(self.created_by_id) + return self.instance_created_by + + # The staff who last edited the task. + def edited_by(self): + if not self.instance_edited_by: + self.instance_edited_by = Location.find(self.edited_by_id) + return self.instance_edited_by + \ No newline at end of file diff --git a/mosaiq/field.py b/mosaiq/field.py index 2eeee09..89da3de 100644 --- a/mosaiq/field.py +++ b/mosaiq/field.py @@ -15,6 +15,7 @@ from .control_point import ControlPoint from .database import Database from .scheduled_field import ScheduledField +from .delivered_dose import DeliveredDose class Field: diff --git a/mosaiq/location.py b/mosaiq/location.py index 1ba1fff..763381e 100644 --- a/mosaiq/location.py +++ b/mosaiq/location.py @@ -13,6 +13,7 @@ #from tkinter import messagebox from .database import Database +from pprint import pprint class Location: @@ -21,6 +22,17 @@ class Location: def find(cls, id): instance = None row = Database.fetch_one("SELECT * FROM Staff WHERE Staff_ID = '{}'".format(str(id))) + #pprint(row) + if row != None: + instance = cls(row) + return instance + + # Returns a single location (Staff/Machine) matching the given username (User_Name) (or nil if no match). + @classmethod + def find_by_username(cls, user_name): + instance = None + row = Database.fetch_one("SELECT * FROM Staff WHERE User_Name = '{}'".format(str(user_name))) + #pprint(row) if row != None: instance = cls(row) return instance diff --git a/mosaiq/pay_pay.py b/mosaiq/pay_pay.py new file mode 100644 index 0000000..25363e3 --- /dev/null +++ b/mosaiq/pay_pay.py @@ -0,0 +1,38 @@ +# encoding: utf8 + +# A class for reading patient payer data from the Mosaiq database. +# +# Authors: +# Ben George +# +# Python 3.6 + +from .database import Database + +from pprint import pprint + +class PatPay: + + # Returns a single task matching the given database id (Payer_ID) (or None if no match). + @classmethod + def find(cls, id): + instance = None + row = Database.fetch_one("SELECT * FROM Pat_Pay WHERE PP_ID = '{}'".format(str(id))) + if row != None: + instance = cls(row) + return instance + + # Returns the payer for a given patient. + @classmethod + def for_patient(cls, patient): + instance = None + row = Database.fetch_one("SELECT * FROM Pat_Pay WHERE Pat_ID1 = '{}'".format(str(patient.id))) + if row != None: + instance = cls(row) + return instance + + # Creates a Payer instance from a payer database row. + def __init__(self, row): + # Database attributes: + self.pat_pay_id = row['PP_ID'] + self.payer_id = row['Payer_ID'] \ No newline at end of file diff --git a/mosaiq/payer.py b/mosaiq/payer.py new file mode 100644 index 0000000..ca90817 --- /dev/null +++ b/mosaiq/payer.py @@ -0,0 +1,39 @@ +# encoding: utf8 + +# A class for reading Episode of Care data from the Mosaiq database. +# +# Authors: +# Ben George +# +# Python 3.6 + +from .database import Database + +from pprint import pprint + +class Payer: + + # Returns a single task matching the given database id (Payer_ID) (or None if no match). + @classmethod + def find(cls, id): + instance = None + row = Database.fetch_one("SELECT * FROM Payer WHERE Payer_ID = '{}'".format(str(id))) + if row != None: + instance = cls(row) + return instance + + @classmethod + def find_all(cls): + instance = None + rows = Database.fetch_all("SELECT * FROM Payer") + payers = list() + for row in rows: + payers.append(cls(row)) + payers.sort(key=lambda p: p.payer_id, reverse=False) + return payers + + # Creates a Payer instance from a payer database row. + def __init__(self, row): + # Database attributes: + self.payer_id = row['Payer_ID'] + self.payer_name = row['Payer_Name'] \ No newline at end of file diff --git a/mosaiq/site_setup.py b/mosaiq/site_setup.py index c0ab6ef..bb1b3a4 100644 --- a/mosaiq/site_setup.py +++ b/mosaiq/site_setup.py @@ -50,9 +50,9 @@ def __init__(self, row): self.patient_orientation_id = row['Patient_Orient'] self.prescribed_offset_id = row['Off_Set_ID'] self.description = row['Setup_Technique_Description'] - self.iso_x = float(row['Isocenter_Position_X']) - self.iso_y = float(row['Isocenter_Position_Y']) - self.iso_z = float(row['Isocenter_Position_Z']) + #self.iso_x = float(row['Isocenter_Position_X']) + #self.iso_y = float(row['Isocenter_Position_Y']) + #self.iso_z = float(row['Isocenter_Position_Z']) self.approved_date = row['Sanct_DtTm'] self.approved_by_id = row['Sanct_ID'] self.status_id = row['Status_Enum']