Added basic Mosaiq interface for payer and episode of care.

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
+from .task import Task
+from .episode import Episode
+from .payer import Payer
+from .pay_pay import PatPay

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']

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,14 +29,39 @@ 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).
   # Returns appointments sorted by their start_date parameter.
   @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
 

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:

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()

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

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:

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
+

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: