Module `phc.easy.ocr.document_composition`

Expand source code

from typing import List, Optional

from phc.easy.auth import Auth
from phc.easy.composition import Composition
from phc.easy.query.fhir_dsl_query import foreign_ids_adder, term_adder
from toolz import pipe

PAGE_NUMBER_COLUMN = (
    "meta.tag_system__lifeomic.com/ocr/documents/page-number__code"
)


class DocumentComposition(Composition):
    @classmethod
    def get(
        cls,
        id: str,
        auth_args: Auth = Auth.shared(),
        query_overrides={},
        **kw_args,
    ):
        query_overrides = pipe(
            query_overrides,
            term_adder({"meta.tag.code.keyword": "PrecisionOCR Service"}),
        )

        return (
            super()
            .get_data_frame(
                id=id,
                auth_args=auth_args,
                query_overrides=query_overrides,
                **kw_args,
            )
            .to_dict("records")[0]
        )

    @classmethod
    def get_data_frame(
        cls,
        document_id: Optional[str] = None,
        document_ids: List[str] = [],
        all_results=False,
        auth_args: Auth = Auth.shared(),
        query_overrides={},
        **kw_args,
    ):
        query_overrides = pipe(
            query_overrides,
            term_adder({"meta.tag.code.keyword": "PrecisionOCR Service"}),
            foreign_ids_adder(
                foreign_id=document_id,
                foreign_ids=document_ids,
                foreign_key="relatesTo.targetReference.reference",
                foreign_id_prefixes=["DocumentReference/"],
            ),
        )

        frame = super().get_data_frame(
            all_results=all_results,
            auth_args=auth_args,
            query_overrides=query_overrides,
            **{"ignore_cache": True, **kw_args},
        )

        if PAGE_NUMBER_COLUMN in frame.columns:
            frame = frame.astype({PAGE_NUMBER_COLUMN: "int"})

        if document_id is not None and PAGE_NUMBER_COLUMN in frame.columns:
            return frame.sort_values(PAGE_NUMBER_COLUMN)

        return frame

Classes

class DocumentComposition

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

Expand source code

class DocumentComposition(Composition):
    @classmethod
    def get(
        cls,
        id: str,
        auth_args: Auth = Auth.shared(),
        query_overrides={},
        **kw_args,
    ):
        query_overrides = pipe(
            query_overrides,
            term_adder({"meta.tag.code.keyword": "PrecisionOCR Service"}),
        )

        return (
            super()
            .get_data_frame(
                id=id,
                auth_args=auth_args,
                query_overrides=query_overrides,
                **kw_args,
            )
            .to_dict("records")[0]
        )

    @classmethod
    def get_data_frame(
        cls,
        document_id: Optional[str] = None,
        document_ids: List[str] = [],
        all_results=False,
        auth_args: Auth = Auth.shared(),
        query_overrides={},
        **kw_args,
    ):
        query_overrides = pipe(
            query_overrides,
            term_adder({"meta.tag.code.keyword": "PrecisionOCR Service"}),
            foreign_ids_adder(
                foreign_id=document_id,
                foreign_ids=document_ids,
                foreign_key="relatesTo.targetReference.reference",
                foreign_id_prefixes=["DocumentReference/"],
            ),
        )

        frame = super().get_data_frame(
            all_results=all_results,
            auth_args=auth_args,
            query_overrides=query_overrides,
            **{"ignore_cache": True, **kw_args},
        )

        if PAGE_NUMBER_COLUMN in frame.columns:
            frame = frame.astype({PAGE_NUMBER_COLUMN: "int"})

        if document_id is not None and PAGE_NUMBER_COLUMN in frame.columns:
            return frame.sort_values(PAGE_NUMBER_COLUMN)

        return frame

Ancestors

Static methods

def code_fields()

Inherited from: Composition.code_fields

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

def get(id: str, auth_args: Auth = <phc.easy.auth.Auth object>, query_overrides={}, **kw_args)

Expand source code

@classmethod
def get(
    cls,
    id: str,
    auth_args: Auth = Auth.shared(),
    query_overrides={},
    **kw_args,
):
    query_overrides = pipe(
        query_overrides,
        term_adder({"meta.tag.code.keyword": "PrecisionOCR Service"}),
    )

    return (
        super()
        .get_data_frame(
            id=id,
            auth_args=auth_args,
            query_overrides=query_overrides,
            **kw_args,
        )
        .to_dict("records")[0]
    )

def get_codes(display_query: Optional[str] = None, sample_size: Optional[int] = None, exclude_meta_tag=True, **kwargs)

Inherited from: Composition.get_codes

Find all codes …

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

Inherited from: Composition.get_count

Get the count for a given FSS query

def get_count_by_field(field: str, **kwargs)

Inherited from: Composition.get_count_by_field

Count records by a given field …

def get_count_by_patient(**kwargs)

Inherited from: Composition.get_count_by_patient

Count records by a given field …

def get_data_frame(document_id: Optional[str] = None, document_ids: List[str] = [], all_results=False, auth_args: Auth = <phc.easy.auth.Auth object>, query_overrides={}, **kw_args)

Inherited from: Composition.get_data_frame

Retrieve records …

Expand source code

@classmethod
def get_data_frame(
    cls,
    document_id: Optional[str] = None,
    document_ids: List[str] = [],
    all_results=False,
    auth_args: Auth = Auth.shared(),
    query_overrides={},
    **kw_args,
):
    query_overrides = pipe(
        query_overrides,
        term_adder({"meta.tag.code.keyword": "PrecisionOCR Service"}),
        foreign_ids_adder(
            foreign_id=document_id,
            foreign_ids=document_ids,
            foreign_key="relatesTo.targetReference.reference",
            foreign_id_prefixes=["DocumentReference/"],
        ),
    )

    frame = super().get_data_frame(
        all_results=all_results,
        auth_args=auth_args,
        query_overrides=query_overrides,
        **{"ignore_cache": True, **kw_args},
    )

    if PAGE_NUMBER_COLUMN in frame.columns:
        frame = frame.astype({PAGE_NUMBER_COLUMN: "int"})

    if document_id is not None and PAGE_NUMBER_COLUMN in frame.columns:
        return frame.sort_values(PAGE_NUMBER_COLUMN)

    return frame

def table_name()

Inherited from: Composition.table_name

Returns the FSS table name for retrieval

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

Inherited from: Composition.transform_results

Transform data frame batch