doczyai-pipelines/fieldExtraction/constants/constants.py

from sentence_transformers import SentenceTransformer
from src.crosswalk.crosswalk_builder import CrosswalkBuilder
from src.crosswalk.list_builder import ListBuilder

from constants.embedding import CodeEmbedding


class Constants:
    """Class to manage all constants and mappings for a Doczy.ai execution"""

    def __init__(self):

        #################################### Dynamic and Exhibit-Level ####################################

        # Dynamic Primary
        self.CROSSWALK_LOB = CrosswalkBuilder().from_json(path="constants/mappings/crosswalk_lob.json")
        self.VALID_LOBS = list(self.CROSSWALK_LOB.mapping.keys())
        
        self.CROSSWALK_PROGRAM = CrosswalkBuilder().from_json(path="constants/mappings/crosswalk_program.json")
        self.VALID_PROGRAMS = list(self.CROSSWALK_PROGRAM.mapping.keys())
        
        self.CROSSWALK_NETWORK = CrosswalkBuilder().from_json(path="constants/mappings/crosswalk_network.json")
        self.VALID_NETWORKS = list(self.CROSSWALK_NETWORK.mapping.keys())
        
        self.CROSSWALK_PRODUCT = CrosswalkBuilder().from_json(path="constants/mappings/crosswalk_product_lob.json")
        self.VALID_PRODUCTS = list(self.CROSSWALK_PRODUCT.mapping.keys())

        # Dynamic Codes
        self.CROSSWALK_PROV_SPECIALTY_CD = CrosswalkBuilder().from_json(path="constants/mappings/crosswalk_provider_specialty.json")
        self.VALID_PROV_SPECIALTY = list(self.CROSSWALK_PROV_SPECIALTY_CD.mapping.keys())
        
        self.CROSSWALK_BILL_TYPE_CD = CrosswalkBuilder().from_json(path="constants/mappings/crosswalk_bill_type.json")
        self.VALID_BILL_TYPE = list(set(self.CROSSWALK_BILL_TYPE_CD.mapping.values()))
        
        self.CROSSWALK_PLACE_OF_SERVICE_CD = CrosswalkBuilder().from_json(path="constants/mappings/crosswalk_place_of_service.json")
        self.VALID_PLACE_OF_SERVICE = list(self.CROSSWALK_PLACE_OF_SERVICE_CD.mapping.keys())
        
        self.CROSSWALK_CLAIM_TYPE_CD =CrosswalkBuilder().from_json(path="constants/mappings/crosswalk_claim_type.json")
        self.VALID_CLAIM_TYPE = list(set(self.CROSSWALK_BILL_TYPE_CD.mapping.keys()))

        # Provider Type
        self.LIST_VALID_PROV_TYPE = ListBuilder().from_json("constants/lists/valid_prov_type.json").list

        #################################### Code Breakout ####################################
        self.SYNONYM_MAP = (
            CrosswalkBuilder()
            .from_json(path="constants/mappings/synonym_map.json")
            .mapping
        )
        self.STOP_WORD_LIST = (
            ListBuilder().from_json("constants/lists/stop_words.json").list
        )
        self.DO_NOT_RUN = (
            ListBuilder().from_json("constants/lists/do_not_run.json").list
        )
        self.REMOVAL_LIST = self._load_removal_list()

        # Code Crosswalks
        self.CPT_LEVEL1_MAPPING = (
            CrosswalkBuilder()
            .from_excel(
                "constants/mapping_csvs/proc_cd/cpt_level1.csv", "Code", "Description"
            )
            .mapping
        )
        self.CPT_LEVEL2_MAPPING = (
            CrosswalkBuilder()
            .from_excel(
                "constants/mapping_csvs/proc_cd/cpt_level2.csv", "Code", "Description"
            )
            .mapping
        )
        # self.CPT_LEVEL3_MAPPING = CrosswalkBuilder().from_excel("constants/mapping_csvs/proc_cd/cpt_level3.csv", "Code", "Description").mapping

        self.HCPCS_LEVEL1_MAPPING = (
            CrosswalkBuilder()
            .from_excel(
                "constants/mapping_csvs/proc_cd/hcpcs_level1.csv", "Code", "Description"
            )
            .mapping
        )
        self.HCPCS_LEVEL2_MAPPING = (
            CrosswalkBuilder()
            .from_excel(
                "constants/mapping_csvs/proc_cd/hcpcs_level2.csv", "Code", "Description"
            )
            .mapping
        )
        # self.HCPCS_MAPPING = CrosswalkBuilder().from_excel("constants/mapping_csvs/proc_cd/hcpcs.csv", "Code", "Description").mapping

        self.REV_LEVEL1_MAPPING = (
            CrosswalkBuilder()
            .from_excel(
                "constants/mapping_csvs/rev_cd/rev_level1.csv", "Code", "Description"
            )
            .mapping
        )
        self.REV_MAPPING = (
            CrosswalkBuilder()
            .from_excel("constants/mapping_csvs/rev_cd/rev.csv", "Code", "Description")
            .mapping
        )
        self.GROUPER_APR_DRG_MAPPING = (
            CrosswalkBuilder()
            .from_excel("constants/mapping_csvs/grouper_cd/drg_mapping_apr-drg.csv", "Code", "Description")
            .mapping
        )
        self.GROUPER_MS_DRG_MAPPING = (
            CrosswalkBuilder()
            .from_excel("constants/mapping_csvs/grouper_cd/drg_mapping_ms-drg.csv", "Code", "Description")
            .mapping
        )

        self.BILL_TYPE_MAPPING = (
            CrosswalkBuilder()
            .from_json("constants/mappings/crosswalk_bill_type.json")
            .mapping
        )
        self.BILL_TYPE_REVERSE_MAPPING = (
            CrosswalkBuilder()
            .from_json("constants/mappings/crosswalk_bill_type.json")
            .create_reverse_mapping()
        )

        # Embedding Model
        self.EMBEDDING_MODEL = SentenceTransformer("all-roberta-large-v1")

        # Embeddings
        self.CPT_LEVEL1_EMBEDDING = CodeEmbedding("cpt_level1")
        self.CPT_LEVEL2_EMBEDDING = CodeEmbedding("cpt_level2")
        # self.CPT_LEVEL3_EMBEDDING = CodeEmbedding("cpt_level3")

        self.HCPCS_LEVEL1_EMBEDDING = CodeEmbedding("hcpcs_level1")
        self.HCPCS_LEVEL2_EMBEDDING = CodeEmbedding("hcpcs_level2")

        self.REV_LEVEL1_EMBEDDING = CodeEmbedding("rev_level1")
        self.REV_EMBEDDING = CodeEmbedding("rev")

        #################################### Methodology Breakout ####################################
        self.VALID_AARETE_DERIVED_FEE_SCHEDULE = (
            ListBuilder().from_json(path="constants/lists/valid_fee_schedule.json").list
        )
        self.VALID_AARETE_DERIVED_FEE_SCHEDULE_VERSION = (
            ListBuilder()
            .from_json(path="constants/lists/valid_fee_schedule_version.json")
            .list
        )
        self.VALID_REIMB_METHODOLOGY = (
            ListBuilder()
            .from_json(path="constants/lists/valid_reimb_methodology.json")
            .list
        )
        self.VALID_UNIT_OF_MEASURE = (
            ListBuilder()
            .from_json(path="constants/lists/valid_unit_of_measure.json")
            .list
        )

        self.VALID_CARVEOUTS = (
            CrosswalkBuilder()
            .from_json(path="constants/mappings/valid_carveouts.json")
            .mapping
        )

        #################################### Other ####################################
        self.COMPOUND_INDICATORS = (
            ListBuilder()
            .from_json(path="constants/lists/compound_indicators.json")
            .list
        )
        self.EXHIBIT_HEADER_MARKERS = (
            ListBuilder()
            .from_json(path="constants/lists/exhibit_header_markers.json")
            .list
        )

    def _load_removal_list(self):
        return [
            v.upper()
            for v in self.VALID_LOBS
            + self.VALID_PRODUCTS
            + list(
                set(
                    CrosswalkBuilder()
                    .from_json(path="constants/mappings/crosswalk_network.json")
                    .mapping.values()
                )
            )
            + self.VALID_CLAIM_TYPE
            + list(
                set(
                    CrosswalkBuilder()
                    .from_json(path="constants/mappings/crosswalk_program.json")
                    .mapping.values()
                )
            )
            + ["INPATIENT", "OUTPATIENT", "IN-PATIENT", "OUT-PATIENT", "IP", "OP"]
            + [
                "COVERED SERVICES",
                "SERVICES",
                "PROCEDURES",
                "CONTRACTED",
                "HEALTHCARE",
                "HEALTH CARE",
                "ALL OTHER",
                "PER DIEM",
                "ADMISSION",
                "REIMBURSEMENT",
                "ADDITIONAL PAYMENT",
                "PAYMENT",
                "RATE",
                "FEE SCHEDULE",
                "METHODOLOGY",
                "OUTLIER",
                "INTERMEDIATE",
                "READMISSION",
                "ACUTE",
                "OPPS",
                "FEE-FOR-SERVICE",
                "PROGRAM",
                "PRODUCT",
            ]  # Generic Terms not found in any of the proc code mappings
        ]

    def get_embedding(self, level: str):
        """
        Returns the CodeEmbedding attribute corresponding to the given level.

        Args:
            level (str): The embedding level identifier (e.g., "rev", "cpt_level1").

        Returns:
            CodeEmbedding: The matching embedding object, or None if not found.
        """
        level_map = {
            # Revenue code embeddings
            "rev": self.REV_EMBEDDING,
            "rev_level1": self.REV_LEVEL1_EMBEDDING,
            # CPT code embeddings
            "cpt_level1": self.CPT_LEVEL1_EMBEDDING,
            "cpt_level2": self.CPT_LEVEL2_EMBEDDING,
            # "cpt_level3": self.CPT_LEVEL3_EMBEDDING,  # Uncomment if needed
            # HCPCS code embeddings
            "hcpcs_level1": self.HCPCS_LEVEL1_EMBEDDING,
            "hcpcs_level2": self.HCPCS_LEVEL2_EMBEDDING,
        }

        return level_map.get(level)

    def get_constant(self, attr_name: str):
        """
        Returns the value of a constant attribute by its string name.
        
        Args:
            attr_name (str): The name of the constant attribute to retrieve.
            
        Returns:
            The value of the requested attribute, or None if the attribute doesn't exist.
            
        Example:
            constants.get_constant("VALID_BILL_TYPE") will return the VALID_BILL_TYPE list.
        """
        if not isinstance(attr_name, str):
            return None  
        if hasattr(self, attr_name):
            return getattr(self, attr_name)
        return None