Module `phc.easy.observation`

Expand source code

from typing import Optional

import pandas as pd
from phc.easy.util.frame import search
from phc.easy.summary.item_counts import SummaryItemCounts
from phc.easy.abstract.fhir_service_patient_item import FhirServicePatientItem
from phc.easy.frame import Frame


class Observation(FhirServicePatientItem):
    @staticmethod
    def table_name():
        return "observation"

    @staticmethod
    def code_fields():
        return [
            "meta.tag",
            "code.coding",
            "component.code.coding",
            "valueCodeableConcept.coding",
            "category.coding",
            "referenceRange.type.coding",
        ]

    @classmethod
    def get_codes(cls, query: Optional[str] = None):
        """Find codes based on case-insensitive matching of code/display/system

        Example
        --------
        >>> import phc.easy as phc
        >>> phc.Auth.set({'account': '<your-account-name>'})
        >>> phc.Project.set_current('My Project Name')
        >>>
        >>> phc.Observation.get_codes("loinc")
        """
        return search(
            SummaryItemCounts.get_data_frame(cls.table_name()), query=query
        )

    @staticmethod
    def transform_results(data_frame: pd.DataFrame, **expand_args):
        args = {
            **expand_args,
            "code_columns": [
                *expand_args.get("code_columns", []),
                "component",
                "interpretation",
            ],
            "custom_columns": [
                *expand_args.get("custom_columns", []),
                Frame.codeable_like_column_expander("subject"),
                Frame.codeable_like_column_expander("related"),
                Frame.codeable_like_column_expander("performer"),
                Frame.codeable_like_column_expander("context"),
            ],
        }

        return Frame.expand(data_frame, **args)

Classes

class Observation

Provides an abstract class and/or static methods for retrieving items from a FSS table that relates to a patient

Expand source code

class Observation(FhirServicePatientItem):
    @staticmethod
    def table_name():
        return "observation"

    @staticmethod
    def code_fields():
        return [
            "meta.tag",
            "code.coding",
            "component.code.coding",
            "valueCodeableConcept.coding",
            "category.coding",
            "referenceRange.type.coding",
        ]

    @classmethod
    def get_codes(cls, query: Optional[str] = None):
        """Find codes based on case-insensitive matching of code/display/system

        Example
        --------
        >>> import phc.easy as phc
        >>> phc.Auth.set({'account': '<your-account-name>'})
        >>> phc.Project.set_current('My Project Name')
        >>>
        >>> phc.Observation.get_codes("loinc")
        """
        return search(
            SummaryItemCounts.get_data_frame(cls.table_name()), query=query
        )

    @staticmethod
    def transform_results(data_frame: pd.DataFrame, **expand_args):
        args = {
            **expand_args,
            "code_columns": [
                *expand_args.get("code_columns", []),
                "component",
                "interpretation",
            ],
            "custom_columns": [
                *expand_args.get("custom_columns", []),
                Frame.codeable_like_column_expander("subject"),
                Frame.codeable_like_column_expander("related"),
                Frame.codeable_like_column_expander("performer"),
                Frame.codeable_like_column_expander("context"),
            ],
        }

        return Frame.expand(data_frame, **args)

Ancestors

Static methods

def code_fields()

Inherited from: FhirServicePatientItem.code_fields

Returns the code keys (e.g. when searching for codes)

Expand source code

@staticmethod
def code_fields():
    return [
        "meta.tag",
        "code.coding",
        "component.code.coding",
        "valueCodeableConcept.coding",
        "category.coding",
        "referenceRange.type.coding",
    ]

def get_codes(query: Optional[str] = None)

Find codes based on case-insensitive matching of code/display/system

Example

>>> import phc.easy as phc
>>> phc.Auth.set({'account': '<your-account-name>'})
>>> phc.Project.set_current('My Project Name')
>>>
>>> phc.Observation.get_codes("loinc")

Expand source code

@classmethod
def get_codes(cls, query: Optional[str] = None):
    """Find codes based on case-insensitive matching of code/display/system

    Example
    --------
    >>> import phc.easy as phc
    >>> phc.Auth.set({'account': '<your-account-name>'})
    >>> phc.Project.set_current('My Project Name')
    >>>
    >>> phc.Observation.get_codes("loinc")
    """
    return search(
        SummaryItemCounts.get_data_frame(cls.table_name()), query=query
    )

def get_count(query_overrides: dict = {}, auth_args=<phc.easy.auth.Auth object>)

Inherited from: FhirServicePatientItem.get_count

Get the count for a given FSS query

def get_count_by_field(field: str, **kwargs)

Inherited from: FhirServicePatientItem.get_count_by_field

Count records by a given field …

def get_count_by_patient(**kwargs)

Inherited from: FhirServicePatientItem.get_count_by_patient

Count records by a given field …

def get_data_frame(all_results: bool = False, raw: bool = False, id: Optional[str] = None, ids: List[str] = [], patient_id: Optional[str] = None, patient_ids: List[str] = [], device_id: Optional[str] = None, device_ids: List[str] = [], page_size: Optional[int] = None, max_pages: Optional[int] = None, query_overrides: dict = {}, auth_args=<phc.easy.auth.Auth object>, ignore_cache: bool = False, expand_args: dict = {}, log: bool = False, term: Optional[dict] = None, terms: List[dict] = [], max_terms: int = 30000, code: Union[str, List[str], None] = None, display: Union[str, List[str], None] = None, system: Union[str, List[str], None] = None, code_fields: List[str] = [])

Inherited from: FhirServicePatientItem.get_data_frame

Retrieve records …

def table_name()

Inherited from: FhirServicePatientItem.table_name

Returns the FSS table name for retrieval

Expand source code

@staticmethod
def table_name():
    return "observation"

def transform_results(data_frame: pandas.core.frame.DataFrame, **expand_args)

Inherited from: FhirServicePatientItem.transform_results

Transform data frame batch

Expand source code

@staticmethod
def transform_results(data_frame: pd.DataFrame, **expand_args):
    args = {
        **expand_args,
        "code_columns": [
            *expand_args.get("code_columns", []),
            "component",
            "interpretation",
        ],
        "custom_columns": [
            *expand_args.get("custom_columns", []),
            Frame.codeable_like_column_expander("subject"),
            Frame.codeable_like_column_expander("related"),
            Frame.codeable_like_column_expander("performer"),
            Frame.codeable_like_column_expander("context"),
        ],
    }

    return Frame.expand(data_frame, **args)