2745

pending completion

Build # 2745

Build Type

Pull #702

travis-ci

Committed by

web-flow

Commit Message

a light weight change the keeps the sparklines but hides them for the time being and replaces them with 2 extra rows refs #700

Pull Request Pull Request #702: a light weight change the keeps the sparklines but hides them for the…

Run Details

61 of 197 branches covered (30.96%)

1 of 1 new or added line in 1 file covered. (100.0%)

2452 of 3148 relevant lines covered (77.89%)

2.05 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

56.13

/elcid/api.py

import datetime
import re
from operator import itemgetter
from collections import defaultdict
from django.conf import settings
from django.utils.text import slugify
from django.http import HttpResponseBadRequest
from intrahospital_api import loader
from rest_framework import viewsets
from opal.core.api import OPALRouter
from opal.core.api import patient_from_pk, LoginRequiredViewset
from opal.core.views import json_response
from opal.core import serialization
from elcid import models as emodels
from elcid.utils import timing
from opal import models as omodels
import os
import json


_LAB_TEST_TAGS = {
    "BIOCHEMISTRY": [
        "BONE PROFILE",
        "UREA AND ELECTROLYTES",
        "LIVER PROFILE",
        "IMMUNOGLOBULINS",
        "C REACTIVE PROTEIN",
        "RENAL"

    ],
    "HAEMATOLOGY": [
        "FULL BLOOD COUNT", "HAEMATINICS", "HBA1C"
    ],
    "ENDOCRINOLOGY": [
        "CORTISOL AT 9AM", "THYROID FUNCTION TESTS"
    ],
    "URINE": [
        "OSMALITY", "PROTEIN ELECTROPHORESIS"
    ],
}

_ALWAYS_SHOW_AS_TABULAR = [
    "UREA AND ELECTROLYTES",
    "LIVER PROFILE",
    "IMMUNOGLOBULINS",
    "C REACTIVE PROTEIN",
    "RENAL",
    "RENAL PROFILE",
    "BONE PROFILE",
    "FULL BLOOD COUNT",
    "HAEMATINICS",
    "HBA1C",
    "THYROID FUNCTION TESTS",
    "ARTERIAL BLOOD GASES",
    "B12 AND FOLATE SCREEN",
    "CLOTTING SCREEN",
    "BICARBONATE",
    "CARDIAC PROFILE",
    "CHLORIDE",
    "CHOLESTEROL/TRIGLYCERIDES",
    "AFP",
    "25-OH VITAMIN D",
    "AMMONIA",
    "FLUID CA-125",
    "CARDIAC TROPONIN T",
    "BODYFLUID CALCIUM",
    "BODYFLUID GLUCOSE",
    "BODYFLUID POTASSIUM",
    "PDF PROTEIN",
    "TACROLIMUS",
    "FULL BLOOD COUNT"
]

LAB_TEST_TAGS = defaultdict(list)

for tag, test_names in _LAB_TEST_TAGS.items():
    for test_name in test_names:
        LAB_TEST_TAGS[test_name].append(tag)


def generate_time_series(observations):
    """
        take in a bunch of observations and return a list
        of numeric observation values
        If we can't pass the observation to a float, we skip it.
    """
    timeseries = []
    for observation in observations:
        obs_result = get_observation_value(observation)

        if obs_result:
            timeseries.append(obs_result)

    return timeseries


def extract_observation_value(observation_value):
    """
        if an observation is numeric, return it as a float
        if its >12 return >12
        else return None
    """
    regex = r'^[-0-9][0-9.]*$'
    obs_result = observation_value.strip()
    obs_result = obs_result.split("~")[0].strip("<").strip(">").strip()
    if re.match(regex, obs_result):
        return round(float(obs_result), 3)


def get_observation_value(observation):
    return extract_observation_value(observation["observation_value"])


def clean_ref_range(ref_range):
    return ref_range.replace("]", "").replace("[", "").strip()


def to_date_str(some_date):
    if some_date:
        return some_date[:10]


def datetime_to_str(dt):
    return dt.strftime(
        settings.DATETIME_INPUT_FORMATS[0]
    )


def observations_by_date(observations):
    """ takes in a bunch of observations, assumes that
        they are like they are in the model, that
    """
    by_date_str = {}

    date_keys = [o['observation_datetime'] for o in observations]
    date_keys = sorted(date_keys)
    date_keys.reverse()
    date_keys = [to_date_str(d) for d in date_keys]

    for observation in observations:
        by_date_str[
            to_date_str(observation['observation_datetime'])
        ] = observation

    return [by_date_str[date_key] for date_key in date_keys]


def get_reference_range(observation):
    ref_range = clean_ref_range(observation["reference_range"])
    if not len(ref_range.replace("-", "").strip()):
        return None
    range_min_max = ref_range.split("-")
    if len(range_min_max) > 2:
        return None
    return {"min": range_min_max[0].strip(), "max": range_min_max[1].strip()}


AEROBIC = "aerobic"
ANAEROBIC = "anaerobic"


class LabTestResultsView(LoginRequiredViewset):
    """ The Api view of the the results view in the patient detail
        We want to show everything grouped by test, then observation, then
        date.
    """
    base_name = 'lab_test_results_view'

    def aggregate_observations_by_lab_test(self, lab_tests):
        by_test = defaultdict(list)
        # lab tests are sorted by lab test type
        # this is either the lab test type if its a lab test we've
        # defined or its what is in the profile description
        # if its an HL7 jobby
        for lab_test in lab_tests:
            as_dict = lab_test.dict_for_view(None)
            observations = as_dict.get("observations", [])
            lab_test_type = as_dict["extras"].get(
                "test_name", lab_test.lab_test_type
            )
            if lab_test_type == "FULL BLOOD COUNT" and observations:
                print("id {} name {} result {}".format(
                    lab_test.id,
                    observations[0]["observation_name"],
                    observations[0]["observation_value"]
                ))

            for observation in observations:
                obs_result = extract_observation_value(observation["observation_value"])
                if obs_result:
                    observation["observation_value"] = obs_result

                observation["reference_range"] = observation["reference_range"].replace("]", "").replace("[", "")

                if not len(observation["reference_range"].replace("-", "").strip()):
                    observation["reference_range"] = None
                else:
                    range_min_max = observation["reference_range"].split("-")
                    if not range_min_max[0].strip():
                        observation["reference_range"] = None
                    else:
                        if not len(range_min_max) == 2:
                            observation["reference_range"] = None
                            # raise ValueError("unable to properly judge the range")
                        else:
                            observation["reference_range"] = dict(
                                min=float(range_min_max[0].strip()),
                                max=float(range_min_max[1].strip())
                            )

                observation["datetime_ordered"] = lab_test.datetime_ordered
                by_test[lab_test_type].append(observation)

        return by_test

    def is_empty_value(self, observation_value):
        """ we don't care about empty strings or
            ' - ' or ' # '
        """
        if isinstance(observation_value, str):
            return not observation_value.strip().strip("-").strip("#").strip()
        else:
            return observation_value is None

    @patient_from_pk
    def retrieve(self, request, patient):

        # so what I want to return is all observations to lab test desplay
        # name with the lab test properties on the observation

        a_year_ago = datetime.date.today() - datetime.timedelta(365)
        lab_tests = emodels.UpstreamLabTest.objects.filter(patient=patient)
        lab_tests = lab_tests.filter(datetime_ordered__gte=a_year_ago)
        lab_tests = [l for l in lab_tests if l.extras]
        by_test = self.aggregate_observations_by_lab_test(lab_tests)
        serialised_tests = []

        # within the lab test observations should be sorted by test name
        # and within the if we have a date range we want to be exposing
        # them as part of a time series, ie adding in blanks if they
        # aren't populated
        for lab_test_type, observations in by_test.items():
            if lab_test_type.strip().upper() in set([
                'UNPROCESSED SAMPLE COMT', 'COMMENT', 'SAMPLE COMMENT'
            ]):
                continue

            observations = sorted(observations, key=lambda x: x["datetime_ordered"])
            observations = sorted(observations, key=lambda x: x["observation_name"])


            # observation_time_series = defaultdict(list)
            by_observations = defaultdict(list)
            timeseries = {}

            observation_metadata = {}
            observation_date_range = {
                to_date_str(observation["observation_datetime"]) for observation in observations
            }
            observation_date_range = sorted(
                list(observation_date_range),
                key=lambda x: serialization.deserialize_date(x)
            )
            observation_date_range.reverse()
            long_form = False

            for observation in observations:
                test_name = observation["observation_name"]

                if test_name.strip() == "Haem Lab Comment":
                    # skip these for the time being, we may not even
                    # have to bring them in
                    continue

                if test_name.strip() == "Sample Comment":
                    continue

                if lab_test_type in _ALWAYS_SHOW_AS_TABULAR:
                    pass
                else:
                    if isinstance(observation["observation_value"], str):
                        if extract_observation_value(observation["observation_value"].strip(">").strip("<")) is None:
                            long_form = True

                if test_name not in by_observations:
                    obs_for_test_name = {
                        to_date_str(i["observation_datetime"]): i for i in observations if i["observation_name"] == test_name
                    }
                    # if its all None's for a certain observation name lets skip it
                    # ie if WBC is all None, lets not waste the users' screen space
                    # sometimes they just have '-' so lets skip these too
                    if not all(i for i in obs_for_test_name.values() if self.is_empty_value(i)):
                        continue
                    by_observations[test_name] = obs_for_test_name

                if test_name not in observation_metadata:
                    observation_metadata[test_name] = dict(
                        units=observation["units"],
                        reference_range=observation["reference_range"],
                        api_name=slugify(observation["observation_name"])
                    )

            serialised_lab_test = dict(
                long_form=long_form,
                timeseries=timeseries,
                api_name=slugify(lab_test_type),
                observation_metadata=observation_metadata,
                lab_test_type=lab_test_type,
                observations=observations,
                observation_date_range=observation_date_range,
                # observation_time_series=observation_time_series,
                by_observations=by_observations,
                observation_names=sorted(by_observations.keys()),
                tags=LAB_TEST_TAGS.get(lab_test_type, [])
            )
            serialised_tests.append(serialised_lab_test)

            serialised_tests = sorted(
                serialised_tests, key=itemgetter("lab_test_type")
            )

            # ordered by most recent observations first please
            serialised_tests = sorted(
                serialised_tests, key=lambda t: -serialization.deserialize_date(
                    t["observation_date_range"][0]
                ).toordinal()
            )

        all_tags = list(_LAB_TEST_TAGS.keys())

        return json_response(
            dict(
                tags=all_tags,
                tests=serialised_tests
            )
        )


class LabTestSummaryApi(LoginRequiredViewset):
    """
        The API View used in the card list. Returns the last 3 months (approixmately)
        of the tests we care about the most.
    """
    base_name = 'lab_test_summary_api'

    PREFERRED_ORDER = [
        "WBC",
        "Lymphocytes",
        "Neutrophils",
        "INR",
        "C Reactive Protein",
        "ALT",
        "AST",
        "Alkaline Phosphatase"
    ]

    def sort_observations(self, obv):
        if obv["name"] in self.PREFERRED_ORDER:
            return self.PREFERRED_ORDER.index(obv["name"])
        else:
            return len(self.PREFERRED_ORDER)

    def aggregate_observations(self, patient):
        """
            takes the specific test/observations
            aggregates them by lab test, observation name
            adds the lab test datetime ordered to the observation dict
            sorts the observations by datetime ordered
        """
        test_data = emodels.UpstreamLabTest.get_relevant_tests(patient)
        result = defaultdict(lambda: defaultdict(list))
        relevant_tests = {
            "C REACTIVE PROTEIN": ["C Reactive Protein"],
            "FULL BLOOD COUNT": ["WBC", "Lymphocytes", "Neutrophils"],
            "LIVER PROFILE": ["ALT", "AST", "Alkaline Phosphatase"],
            "CLOTTING SCREEN": ["INR"]
        }

        for test in test_data:
            test_name = test.extras.get("test_name")
            if test_name in relevant_tests:
                relevent_observations = relevant_tests[test_name]

                for observation in test.extras["observations"]:
                    observation_name = observation["observation_name"]
                    if observation_name in relevent_observations:
                        observation["datetime_ordered"] = test.datetime_ordered
                        result[test_name][observation_name].append(observation)

        for test, obs_collection in result.items():
            for obs_name, obs in obs_collection.items():
                result[test][obs_name] = observations_by_date(obs)
        return result

    @timing
    def serialise_lab_tests(self, patient):
        aggregated_data = self.aggregate_observations(patient)

        serialised_obvs = []
        all_dates = set()

        for test_name, obs_collection in aggregated_data.items():
            for observation_name, observations in obs_collection.items():
                serialised_obvs.append(dict(
                    name=observation_name,
                    graph_values=generate_time_series(observations),
                    units=observations[0]["units"],
                    reference_range=get_reference_range(observations[0]),
                    latest_results={
                        to_date_str(datetime_to_str(i["datetime_ordered"])): get_observation_value(i) for i in observations if get_observation_value(i)
                    }
                ))
                all_dates = all_dates.union(
                    i["datetime_ordered"].date() for i in observations if get_observation_value(i)
                )

        recent_dates = sorted(list(all_dates))
        recent_dates.reverse()
        recent_dates = recent_dates[:5]
        obs_values = sorted(serialised_obvs, key=self.sort_observations)
        result = dict(
            obs_values=obs_values,
            recent_dates=recent_dates
        )

        return result

    @patient_from_pk
    def retrieve(self, request, patient):
        result = self.serialise_lab_tests(patient)
        return json_response(result)


class UpstreamBloodCultureApi(viewsets.ViewSet):
    """
        for every upstream blood culture, return them grouped by

        datetimes_ordered as a date,
        lab_number
        observation name
    """
    base_name = "upstream_blood_culture_results"

    def no_growth_observations(self, observations):
        """
            We are looking for observations that looks like the below.
            Its not necessarily 5 days, sometimes its e.g. 48 hours.
            Otherwise they always look like the below.

            Aerobic bottle culture: No growth after 5 days of incubation
            Anaerobic bottle culture: No growth after 5 days of incubation

            The are always of the type [Anaerobic/Aerobic] bottle culture
        """
        obs_names = ["Aerobic bottle culture", "Anaerobic bottle culture"]

        bottles = [
            o for o in observations if o["observation_name"] in obs_names
        ]

        return len(bottles) == 2

    @patient_from_pk
    def retrieve(self, request, patient):
        """
            returns any observations with Aerobic or Anaerobic in their name
        """
        lab_tests = patient.labtest_set.filter(
            lab_test_type=emodels.UpstreamBloodCulture.get_display_name()
        ).order_by("external_identifier").order_by("-datetime_ordered")
        lab_tests = [i.dict_for_view(request.user) for i in lab_tests]
        for lab_test in lab_tests:
            observations = []
            lab_test["no_growth"] = self.no_growth_observations(
                lab_test["observations"]
            )

            for observation in lab_test["observations"]:
                ob_name = observation["observation_name"].lower()

                if "aerobic" in ob_name:
                    observations.append(observation)

            lab_test["observations"] = sorted(
                observations, key=lambda x: x["observation_name"]
            )
            if lab_test["extras"]["clinical_info"]:
                lab_test["extras"]["clinical_info"] = "{}{}".format(
                    lab_test["extras"]["clinical_info"][0].upper(),
                    lab_test["extras"]["clinical_info"][1:]
                )

        return json_response(dict(
            lab_tests=lab_tests
        ))


class BloodCultureResultApi(viewsets.ViewSet):
    base_name = 'blood_culture_results'

    BLOOD_CULTURES = [
        emodels.GramStain.get_display_name(),
        emodels.QuickFISH.get_display_name(),
        emodels.GPCStaph.get_display_name(),
        emodels.GPCStrep.get_display_name(),
        emodels.GNR.get_display_name(),
        emodels.BloodCultureOrganism.get_display_name()
    ]

    def sort_by_date_ordered_and_lab_number(self, some_keys):
        """ takes in a tuple of (date, lab number)
            both date or lab number will be "" if empty

            it sorts them by most recent and lowest lab number
        """
        def comparator(some_key):
            dt = some_key[0]
            if not dt:
                dt = datetime.date.max
            return (dt, some_key[1],)
        return sorted(some_keys, key=comparator, reverse=True)

    def sort_by_lab_test_order(self, some_results):
        """
            results should be sorted by the order of the blood culture
            list
        """
        return sorted(
            some_results, key=lambda x: self.BLOOD_CULTURES.index(
                x["lab_test_type"]
            )
        )

    def translate_date_to_string(self, some_date):
        if not some_date:
            return ""

        dt = datetime.datetime(
            some_date.year, some_date.month, some_date.day
        )
        return dt.strftime(
            settings.DATE_INPUT_FORMATS[0]
        )

    @patient_from_pk
    def retrieve(self, request, patient):
        lab_tests = patient.labtest_set.filter(
            lab_test_type__in=self.BLOOD_CULTURES
        )
        lab_tests = lab_tests.order_by("datetime_ordered")
        cultures = defaultdict(lambda: defaultdict(dict))
        culture_order = set()

        for lab_test in lab_tests:
            lab_number = lab_test.extras.get("lab_number", "")
            # if lab number is None or "", group it together
            if not lab_number:
                lab_number = ""
            if lab_test.datetime_ordered:
                date_ordered = self.translate_date_to_string(
                    lab_test.datetime_ordered.date()
                )

                culture_order.add((
                    lab_test.datetime_ordered.date(), lab_number,
                ))
            else:
                date_ordered = ""
                culture_order.add(("", lab_number,))

            if lab_number not in cultures[date_ordered]:
                cultures[date_ordered][lab_number][AEROBIC] = defaultdict(list)
                cultures[date_ordered][lab_number][ANAEROBIC] = defaultdict(
                    list
                )

            isolate = lab_test.extras["isolate"]

            if lab_test.extras[AEROBIC]:
                cultures[date_ordered][lab_number][AEROBIC][isolate].append(
                    lab_test.to_dict(self.request.user)
                )
            else:
                cultures[date_ordered][lab_number][ANAEROBIC][isolate].append(
                    lab_test.to_dict(request.user)
                )

        culture_order = self.sort_by_date_ordered_and_lab_number(culture_order)

        for dt, lab_number in culture_order:
            dt_string = self.translate_date_to_string(dt)
            by_date_lab_number = cultures[dt_string][lab_number]
            for robic in [AEROBIC, ANAEROBIC]:
                for isolate in by_date_lab_number[robic].keys():
                    by_date_lab_number[robic][isolate] = self.sort_by_lab_test_order(
                        by_date_lab_number[robic][isolate]
                    )

        return json_response(dict(
            cultures=cultures,
            culture_order=culture_order
        ))


class DemographicsSearch(LoginRequiredViewset):
    base_name = 'demographics_search'
    PATIENT_FOUND_IN_ELCID = "patient_found_in_elcid"
    PATIENT_FOUND_UPSTREAM = "patient_found_upstream"
    PATIENT_NOT_FOUND = "patient_not_found"

    def list(self, request, *args, **kwargs):
        hospital_number = request.query_params.get("hospital_number")
        if not hospital_number:
            return HttpResponseBadRequest("Please pass in a hospital number")
        demographics = emodels.Demographics.objects.filter(
            hospital_number=hospital_number
        ).last()

        # the patient is in elcid
        if demographics:
            return json_response(dict(
                patient=demographics.patient.to_dict(request.user),
                status=self.PATIENT_FOUND_IN_ELCID
            ))
        else:
            if settings.USE_UPSTREAM_DEMOGRAPHICS:
                demographics = loader.load_demographics(hospital_number)

                if demographics:
                    return json_response(dict(
                        patient=dict(demographics=[demographics]),
                        status=self.PATIENT_FOUND_UPSTREAM
                    ))
        return json_response(dict(status=self.PATIENT_NOT_FOUND))


elcid_router = OPALRouter()
elcid_router.register(BloodCultureResultApi.base_name, BloodCultureResultApi)
elcid_router.register(
    UpstreamBloodCultureApi.base_name, UpstreamBloodCultureApi
)
elcid_router.register(DemographicsSearch.base_name, DemographicsSearch)

lab_test_router = OPALRouter()
lab_test_router.register('lab_test_summary_api', LabTestSummaryApi)
lab_test_router.register('lab_test_results_view', LabTestResultsView)

1	import datetime	1✔
2	import re	1✔
3	from operator import itemgetter	1✔
4	from collections import defaultdict	1✔
5	from django.conf import settings	1✔
6	from django.utils.text import slugify	1✔
7	from django.http import HttpResponseBadRequest	1✔
8	from intrahospital_api import loader	1✔
9	from rest_framework import viewsets	1✔
10	from opal.core.api import OPALRouter	1✔
11	from opal.core.api import patient_from_pk, LoginRequiredViewset	1✔
12	from opal.core.views import json_response	1✔
13	from opal.core import serialization	1✔
14	from elcid import models as emodels	1✔
15	from elcid.utils import timing	1✔
16	from opal import models as omodels	1✔
17	import os	1✔
18	import json	1✔
19
20
21	_LAB_TEST_TAGS = {	1✔
22	"BIOCHEMISTRY": [
23	"BONE PROFILE",
24	"UREA AND ELECTROLYTES",
25	"LIVER PROFILE",
26	"IMMUNOGLOBULINS",
27	"C REACTIVE PROTEIN",
28	"RENAL"
29
30	],
31	"HAEMATOLOGY": [
32	"FULL BLOOD COUNT", "HAEMATINICS", "HBA1C"
33	],
34	"ENDOCRINOLOGY": [
35	"CORTISOL AT 9AM", "THYROID FUNCTION TESTS"
36	],
37	"URINE": [
38	"OSMALITY", "PROTEIN ELECTROPHORESIS"
39	],
40	}
41
42	_ALWAYS_SHOW_AS_TABULAR = [	1✔
43	"UREA AND ELECTROLYTES",
44	"LIVER PROFILE",
45	"IMMUNOGLOBULINS",
46	"C REACTIVE PROTEIN",
47	"RENAL",
48	"RENAL PROFILE",
49	"BONE PROFILE",
50	"FULL BLOOD COUNT",
51	"HAEMATINICS",
52	"HBA1C",
53	"THYROID FUNCTION TESTS",
54	"ARTERIAL BLOOD GASES",
55	"B12 AND FOLATE SCREEN",
56	"CLOTTING SCREEN",
57	"BICARBONATE",
58	"CARDIAC PROFILE",
59	"CHLORIDE",
60	"CHOLESTEROL/TRIGLYCERIDES",
61	"AFP",
62	"25-OH VITAMIN D",
63	"AMMONIA",
64	"FLUID CA-125",
65	"CARDIAC TROPONIN T",
66	"BODYFLUID CALCIUM",
67	"BODYFLUID GLUCOSE",
68	"BODYFLUID POTASSIUM",
69	"PDF PROTEIN",
70	"TACROLIMUS",
71	"FULL BLOOD COUNT"
72	]
73
74	LAB_TEST_TAGS = defaultdict(list)	1✔
75
76	for tag, test_names in _LAB_TEST_TAGS.items():	1✔
77	for test_name in test_names:	1✔
78	LAB_TEST_TAGS[test_name].append(tag)	1✔
79
80
81	def generate_time_series(observations):	1✔
82	"""
83	take in a bunch of observations and return a list
84	of numeric observation values
85	If we can't pass the observation to a float, we skip it.
86	"""
87	timeseries = []	1✔
88	for observation in observations:	1✔
89	obs_result = get_observation_value(observation)	1✔
90
91	if obs_result:	1✔
92	timeseries.append(obs_result)	1✔
93
94	return timeseries	1✔
95
96
97	def extract_observation_value(observation_value):	1✔
98	"""
99	if an observation is numeric, return it as a float
100	if its >12 return >12
101	else return None
102	"""
103	regex = r'^[-0-9][0-9.]*$'	1✔
104	obs_result = observation_value.strip()	1✔
105	obs_result = obs_result.split("~")[0].strip("<").strip(">").strip()	1✔
106	if re.match(regex, obs_result):	1✔
107	return round(float(obs_result), 3)	1✔
108
109
110	def get_observation_value(observation):	1✔
111	return extract_observation_value(observation["observation_value"])	1✔
112
113
114	def clean_ref_range(ref_range):	1✔
115	return ref_range.replace("]", "").replace("[", "").strip()	1✔
116
117
118	def to_date_str(some_date):	1✔
119	if some_date:	1✔
120	return some_date[:10]	1✔
121
122
123	def datetime_to_str(dt):	1✔
124	return dt.strftime(	1✔
125	settings.DATETIME_INPUT_FORMATS[0]
126	)
127
128
129	def observations_by_date(observations):	1✔
130	""" takes in a bunch of observations, assumes that
131	they are like they are in the model, that
132	"""
133	by_date_str = {}	1✔
134
135	date_keys = [o['observation_datetime'] for o in observations]	1✔
136	date_keys = sorted(date_keys)	1✔
137	date_keys.reverse()	1✔
138	date_keys = [to_date_str(d) for d in date_keys]	1✔
139
140	for observation in observations:	1✔
141	by_date_str[	1✔
142	to_date_str(observation['observation_datetime'])
143	] = observation
144
145	return [by_date_str[date_key] for date_key in date_keys]	1✔
146
147
148	def get_reference_range(observation):	1✔
149	ref_range = clean_ref_range(observation["reference_range"])	1✔
150	if not len(ref_range.replace("-", "").strip()):	1✔
151	return None	1✔
152	range_min_max = ref_range.split("-")	1✔
153	if len(range_min_max) > 2:	1✔
154	return None	1✔
155	return {"min": range_min_max[0].strip(), "max": range_min_max[1].strip()}	1✔
156
157
158	AEROBIC = "aerobic"	1✔
159	ANAEROBIC = "anaerobic"	1✔
160
161
162	class LabTestResultsView(LoginRequiredViewset):	1✔
163	""" The Api view of the the results view in the patient detail
164	We want to show everything grouped by test, then observation, then
165	date.
166	"""
167	base_name = 'lab_test_results_view'	1✔
168
169	def aggregate_observations_by_lab_test(self, lab_tests):	1✔
170	by_test = defaultdict(list)	×
171	# lab tests are sorted by lab test type
172	# this is either the lab test type if its a lab test we've
173	# defined or its what is in the profile description
174	# if its an HL7 jobby
175	for lab_test in lab_tests:	×
176	as_dict = lab_test.dict_for_view(None)	×
177	observations = as_dict.get("observations", [])	×
178	lab_test_type = as_dict["extras"].get(	×
179	"test_name", lab_test.lab_test_type
180	)
181	if lab_test_type == "FULL BLOOD COUNT" and observations:	×
182	print("id {} name {} result {}".format(	×
183	lab_test.id,
184	observations[0]["observation_name"],
185	observations[0]["observation_value"]
186	))
187
188	for observation in observations:	×
189	obs_result = extract_observation_value(observation["observation_value"])	×
190	if obs_result:	×
191	observation["observation_value"] = obs_result	×
192
193	observation["reference_range"] = observation["reference_range"].replace("]", "").replace("[", "")	×
194
195	if not len(observation["reference_range"].replace("-", "").strip()):	×
196	observation["reference_range"] = None	×
197	else:
198	range_min_max = observation["reference_range"].split("-")	×
199	if not range_min_max[0].strip():	×
200	observation["reference_range"] = None	×
201	else:
202	if not len(range_min_max) == 2:	×
203	observation["reference_range"] = None	×
204	# raise ValueError("unable to properly judge the range")
205	else:
206	observation["reference_range"] = dict(	×
207	min=float(range_min_max[0].strip()),
208	max=float(range_min_max[1].strip())
209	)
210
211	observation["datetime_ordered"] = lab_test.datetime_ordered	×
212	by_test[lab_test_type].append(observation)	×
213
214	return by_test	×
215
216	def is_empty_value(self, observation_value):	1✔
217	""" we don't care about empty strings or
218	' - ' or ' # '
219	"""
220	if isinstance(observation_value, str):	×
221	return not observation_value.strip().strip("-").strip("#").strip()	×
222	else:
223	return observation_value is None	×
224
225	@patient_from_pk	1✔
226	def retrieve(self, request, patient):
227
228	# so what I want to return is all observations to lab test desplay
229	# name with the lab test properties on the observation
230
231	a_year_ago = datetime.date.today() - datetime.timedelta(365)	×
232	lab_tests = emodels.UpstreamLabTest.objects.filter(patient=patient)	×
233	lab_tests = lab_tests.filter(datetime_ordered__gte=a_year_ago)	×
234	lab_tests = [l for l in lab_tests if l.extras]	×
235	by_test = self.aggregate_observations_by_lab_test(lab_tests)	×
236	serialised_tests = []	×
237
238	# within the lab test observations should be sorted by test name
239	# and within the if we have a date range we want to be exposing
240	# them as part of a time series, ie adding in blanks if they
241	# aren't populated
242	for lab_test_type, observations in by_test.items():	×
243	if lab_test_type.strip().upper() in set([	×
244	'UNPROCESSED SAMPLE COMT', 'COMMENT', 'SAMPLE COMMENT'
245	]):
246	continue	×
247
248	observations = sorted(observations, key=lambda x: x["datetime_ordered"])	×
249	observations = sorted(observations, key=lambda x: x["observation_name"])	×
250
251
252	# observation_time_series = defaultdict(list)
253	by_observations = defaultdict(list)	×
254	timeseries = {}	×
255
256	observation_metadata = {}	×
257	observation_date_range = {	×
258	to_date_str(observation["observation_datetime"]) for observation in observations
259	}
260	observation_date_range = sorted(	×
261	list(observation_date_range),
262	key=lambda x: serialization.deserialize_date(x)
263	)
264	observation_date_range.reverse()	×
265	long_form = False	×
266
267	for observation in observations:	×
268	test_name = observation["observation_name"]	×
269
270	if test_name.strip() == "Haem Lab Comment":	×
271	# skip these for the time being, we may not even
272	# have to bring them in
273	continue	×
274
275	if test_name.strip() == "Sample Comment":	×
276	continue	×
277
278	if lab_test_type in _ALWAYS_SHOW_AS_TABULAR:	×
279	pass	×
280	else:
281	if isinstance(observation["observation_value"], str):	×
282	if extract_observation_value(observation["observation_value"].strip(">").strip("<")) is None:	×
283	long_form = True	×
284
285	if test_name not in by_observations:	×
286	obs_for_test_name = {	×
287	to_date_str(i["observation_datetime"]): i for i in observations if i["observation_name"] == test_name
288	}
289	# if its all None's for a certain observation name lets skip it
290	# ie if WBC is all None, lets not waste the users' screen space
291	# sometimes they just have '-' so lets skip these too
292	if not all(i for i in obs_for_test_name.values() if self.is_empty_value(i)):	×
293	continue	×
294	by_observations[test_name] = obs_for_test_name	×
295
296	if test_name not in observation_metadata:	×
297	observation_metadata[test_name] = dict(	×
298	units=observation["units"],
299	reference_range=observation["reference_range"],
300	api_name=slugify(observation["observation_name"])
301	)
302
303	serialised_lab_test = dict(	×
304	long_form=long_form,
305	timeseries=timeseries,
306	api_name=slugify(lab_test_type),
307	observation_metadata=observation_metadata,
308	lab_test_type=lab_test_type,
309	observations=observations,
310	observation_date_range=observation_date_range,
311	# observation_time_series=observation_time_series,
312	by_observations=by_observations,
313	observation_names=sorted(by_observations.keys()),
314	tags=LAB_TEST_TAGS.get(lab_test_type, [])
315	)
316	serialised_tests.append(serialised_lab_test)	×
317
318	serialised_tests = sorted(	×
319	serialised_tests, key=itemgetter("lab_test_type")
320	)
321
322	# ordered by most recent observations first please
323	serialised_tests = sorted(	×
324	serialised_tests, key=lambda t: -serialization.deserialize_date(
325	t["observation_date_range"][0]
326	).toordinal()
327	)
328
329	all_tags = list(_LAB_TEST_TAGS.keys())	×
330
331	return json_response(	×
332	dict(
333	tags=all_tags,
334	tests=serialised_tests
335	)
336	)
337
338
339	class LabTestSummaryApi(LoginRequiredViewset):	1✔
340	"""
341	The API View used in the card list. Returns the last 3 months (approixmately)
342	of the tests we care about the most.
343	"""
344	base_name = 'lab_test_summary_api'	1✔
345
346	PREFERRED_ORDER = [	1✔
347	"WBC",
348	"Lymphocytes",
349	"Neutrophils",
350	"INR",
351	"C Reactive Protein",
352	"ALT",
353	"AST",
354	"Alkaline Phosphatase"
355	]
356
357	def sort_observations(self, obv):	1✔
358	if obv["name"] in self.PREFERRED_ORDER:	×
359	return self.PREFERRED_ORDER.index(obv["name"])	×
360	else:
361	return len(self.PREFERRED_ORDER)	×
362
363	def aggregate_observations(self, patient):	1✔
364	"""
365	takes the specific test/observations
366	aggregates them by lab test, observation name
367	adds the lab test datetime ordered to the observation dict
368	sorts the observations by datetime ordered
369	"""
370	test_data = emodels.UpstreamLabTest.get_relevant_tests(patient)	×
371	result = defaultdict(lambda: defaultdict(list))	×
372	relevant_tests = {	×
373	"C REACTIVE PROTEIN": ["C Reactive Protein"],
374	"FULL BLOOD COUNT": ["WBC", "Lymphocytes", "Neutrophils"],
375	"LIVER PROFILE": ["ALT", "AST", "Alkaline Phosphatase"],
376	"CLOTTING SCREEN": ["INR"]
377	}
378
379	for test in test_data:	×
380	test_name = test.extras.get("test_name")	×
381	if test_name in relevant_tests:	×
382	relevent_observations = relevant_tests[test_name]	×
383
384	for observation in test.extras["observations"]:	×
385	observation_name = observation["observation_name"]	×
386	if observation_name in relevent_observations:	×
387	observation["datetime_ordered"] = test.datetime_ordered	×
388	result[test_name][observation_name].append(observation)	×
389
390	for test, obs_collection in result.items():	×
391	for obs_name, obs in obs_collection.items():	×
392	result[test][obs_name] = observations_by_date(obs)	×
393	return result	×
394
395	@timing	1✔
396	def serialise_lab_tests(self, patient):
397	aggregated_data = self.aggregate_observations(patient)	×
398
399	serialised_obvs = []	×
400	all_dates = set()	×
401
402	for test_name, obs_collection in aggregated_data.items():	×
403	for observation_name, observations in obs_collection.items():	×
404	serialised_obvs.append(dict(	×
405	name=observation_name,
406	graph_values=generate_time_series(observations),
407	units=observations[0]["units"],
408	reference_range=get_reference_range(observations[0]),
409	latest_results={
410	to_date_str(datetime_to_str(i["datetime_ordered"])): get_observation_value(i) for i in observations if get_observation_value(i)
411	}
412	))
413	all_dates = all_dates.union(	×
414	i["datetime_ordered"].date() for i in observations if get_observation_value(i)
415	)
416
417	recent_dates = sorted(list(all_dates))	×
418	recent_dates.reverse()	×
419	recent_dates = recent_dates[:5]	×
420	obs_values = sorted(serialised_obvs, key=self.sort_observations)	×
421	result = dict(	×
422	obs_values=obs_values,
423	recent_dates=recent_dates
424	)
425
426	return result	×
427
428	@patient_from_pk	1✔
429	def retrieve(self, request, patient):
430	result = self.serialise_lab_tests(patient)	×
431	return json_response(result)	×
432
433
434	class UpstreamBloodCultureApi(viewsets.ViewSet):	1✔
435	"""
436	for every upstream blood culture, return them grouped by
437
438	datetimes_ordered as a date,
439	lab_number
440	observation name
441	"""
442	base_name = "upstream_blood_culture_results"	1✔
443
444	def no_growth_observations(self, observations):	1✔
445	"""
446	We are looking for observations that looks like the below.
447	Its not necessarily 5 days, sometimes its e.g. 48 hours.
448	Otherwise they always look like the below.
449
450	Aerobic bottle culture: No growth after 5 days of incubation
451	Anaerobic bottle culture: No growth after 5 days of incubation
452
453	The are always of the type [Anaerobic/Aerobic] bottle culture
454	"""
455	obs_names = ["Aerobic bottle culture", "Anaerobic bottle culture"]	1✔
456
457	bottles = [	1✔
458	o for o in observations if o["observation_name"] in obs_names
459	]
460
461	return len(bottles) == 2	1✔
462
463	@patient_from_pk	1✔
464	def retrieve(self, request, patient):
465	"""
466	returns any observations with Aerobic or Anaerobic in their name
467	"""
468	lab_tests = patient.labtest_set.filter(	×
469	lab_test_type=emodels.UpstreamBloodCulture.get_display_name()
470	).order_by("external_identifier").order_by("-datetime_ordered")
471	lab_tests = [i.dict_for_view(request.user) for i in lab_tests]	×
472	for lab_test in lab_tests:	×
473	observations = []	×
474	lab_test["no_growth"] = self.no_growth_observations(	×
475	lab_test["observations"]
476	)
477
478	for observation in lab_test["observations"]:	×
479	ob_name = observation["observation_name"].lower()	×
480
481	if "aerobic" in ob_name:	×
482	observations.append(observation)	×
483
484	lab_test["observations"] = sorted(	×
485	observations, key=lambda x: x["observation_name"]
486	)
487	if lab_test["extras"]["clinical_info"]:	×
488	lab_test["extras"]["clinical_info"] = "{}{}".format(	×
489	lab_test["extras"]["clinical_info"][0].upper(),
490	lab_test["extras"]["clinical_info"][1:]
491	)
492
493	return json_response(dict(	×
494	lab_tests=lab_tests
495	))
496
497
498	class BloodCultureResultApi(viewsets.ViewSet):	1✔
499	base_name = 'blood_culture_results'	1✔
500
501	BLOOD_CULTURES = [	1✔
502	emodels.GramStain.get_display_name(),
503	emodels.QuickFISH.get_display_name(),
504	emodels.GPCStaph.get_display_name(),
505	emodels.GPCStrep.get_display_name(),
506	emodels.GNR.get_display_name(),
507	emodels.BloodCultureOrganism.get_display_name()
508	]
509
510	def sort_by_date_ordered_and_lab_number(self, some_keys):	1✔
511	""" takes in a tuple of (date, lab number)
512	both date or lab number will be "" if empty
513
514	it sorts them by most recent and lowest lab number
515	"""
516	def comparator(some_key):	1✔
517	dt = some_key[0]	1✔
518	if not dt:	1✔
519	dt = datetime.date.max	1✔
520	return (dt, some_key[1],)	1✔
521	return sorted(some_keys, key=comparator, reverse=True)	1✔
522
523	def sort_by_lab_test_order(self, some_results):	1✔
524	"""
525	results should be sorted by the order of the blood culture
526	list
527	"""
528	return sorted(	1✔
529	some_results, key=lambda x: self.BLOOD_CULTURES.index(
530	x["lab_test_type"]
531	)
532	)
533
534	def translate_date_to_string(self, some_date):	1✔
535	if not some_date:	1✔
536	return ""	1✔
537
538	dt = datetime.datetime(	1✔
539	some_date.year, some_date.month, some_date.day
540	)
541	return dt.strftime(	1✔
542	settings.DATE_INPUT_FORMATS[0]
543	)
544
545	@patient_from_pk	1✔
546	def retrieve(self, request, patient):
547	lab_tests = patient.labtest_set.filter(	1✔
548	lab_test_type__in=self.BLOOD_CULTURES
549	)
550	lab_tests = lab_tests.order_by("datetime_ordered")	1✔
551	cultures = defaultdict(lambda: defaultdict(dict))	1✔
552	culture_order = set()	1✔
553
554	for lab_test in lab_tests:	1✔
555	lab_number = lab_test.extras.get("lab_number", "")	1✔
556	# if lab number is None or "", group it together
557	if not lab_number:	1✔
558	lab_number = ""	1✔
559	if lab_test.datetime_ordered:	1✔
560	date_ordered = self.translate_date_to_string(	1✔
561	lab_test.datetime_ordered.date()
562	)
563
564	culture_order.add((	1✔
565	lab_test.datetime_ordered.date(), lab_number,
566	))
567	else:
568	date_ordered = ""	×
569	culture_order.add(("", lab_number,))	×
570
571	if lab_number not in cultures[date_ordered]:	1✔
572	cultures[date_ordered][lab_number][AEROBIC] = defaultdict(list)	1✔
573	cultures[date_ordered][lab_number][ANAEROBIC] = defaultdict(	1✔
574	list
575	)
576
577	isolate = lab_test.extras["isolate"]	1✔
578
579	if lab_test.extras[AEROBIC]:	1✔
580	cultures[date_ordered][lab_number][AEROBIC][isolate].append(	×
581	lab_test.to_dict(self.request.user)
582	)
583	else:
584	cultures[date_ordered][lab_number][ANAEROBIC][isolate].append(	1✔
585	lab_test.to_dict(request.user)
586	)
587
588	culture_order = self.sort_by_date_ordered_and_lab_number(culture_order)	1✔
589
590	for dt, lab_number in culture_order:	1✔
591	dt_string = self.translate_date_to_string(dt)	1✔
592	by_date_lab_number = cultures[dt_string][lab_number]	1✔
593	for robic in [AEROBIC, ANAEROBIC]:	1✔
594	for isolate in by_date_lab_number[robic].keys():	1✔
595	by_date_lab_number[robic][isolate] = self.sort_by_lab_test_order(	1✔
596	by_date_lab_number[robic][isolate]
597	)
598
599	return json_response(dict(	1✔
600	cultures=cultures,
601	culture_order=culture_order
602	))
603
604
605	class DemographicsSearch(LoginRequiredViewset):	1✔
606	base_name = 'demographics_search'	1✔
607	PATIENT_FOUND_IN_ELCID = "patient_found_in_elcid"	1✔
608	PATIENT_FOUND_UPSTREAM = "patient_found_upstream"	1✔
609	PATIENT_NOT_FOUND = "patient_not_found"	1✔
610
611	def list(self, request, args, *kwargs):	1✔
612	hospital_number = request.query_params.get("hospital_number")	1✔
613	if not hospital_number:	1✔
614	return HttpResponseBadRequest("Please pass in a hospital number")	1✔
615	demographics = emodels.Demographics.objects.filter(	1✔
616	hospital_number=hospital_number
617	).last()
618
619	# the patient is in elcid
620	if demographics:	1✔
621	return json_response(dict(	1✔
622	patient=demographics.patient.to_dict(request.user),
623	status=self.PATIENT_FOUND_IN_ELCID
624	))
625	else:
626	if settings.USE_UPSTREAM_DEMOGRAPHICS:	1✔
627	demographics = loader.load_demographics(hospital_number)	1✔
628
629	if demographics:	1✔
630	return json_response(dict(	1✔
631	patient=dict(demographics=[demographics]),
632	status=self.PATIENT_FOUND_UPSTREAM
633	))
634	return json_response(dict(status=self.PATIENT_NOT_FOUND))	1✔
635
636
637	elcid_router = OPALRouter()	1✔
638	elcid_router.register(BloodCultureResultApi.base_name, BloodCultureResultApi)	1✔
639	elcid_router.register(	1✔
640	UpstreamBloodCultureApi.base_name, UpstreamBloodCultureApi
641	)
642	elcid_router.register(DemographicsSearch.base_name, DemographicsSearch)	1✔
643
644	lab_test_router = OPALRouter()	1✔
645	lab_test_router.register('lab_test_summary_api', LabTestSummaryApi)	1✔
646	lab_test_router.register('lab_test_results_view', LabTestResultsView)	1✔

openhealthcare / elcid-rfh / 2745

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous