11.10.26.76

updated the backend_update.ipynb for including lab content
added the article.py for the article classes and instruction
for the instruction classes
Changes to be committed:
modified: article.py
modified: backend_update.ipynb
modified: instruction.py

article.py

@@ -2,13 +2,18 @@ from typing import Optional, List, Dict
 from enum import Enum
 
 from pydantic import BaseModel, Field
-# from langchain. import 
-
+from utility import read_pdf
 
 class Article(BaseModel):
-    id: int
-    title: str
-    content: str
+    file: str|object = Field(None, alias="file")
+    title: Optional[str] = Field(None, alias="title")
+    content: Optional[str]
+
+    def read_article(self):
+        if isinstance(self.file,str):
+            self.content = read_pdf(self.file)
+        else:
+            self.content = self.file.read()
 
 class Table(BaseModel):
     title: str = Field(None, alias="title")
@@ -18,13 +23,28 @@ class Table(BaseModel):
 class Summary(BaseModel):
     tables: Optional[List[Table]] = Field(None, alias="tables")
 
+class Region(Enum):
+    spine = "Spine"
+    extremity = "Extremity"
+    all = "All"
+
 class Assessment(Enum):
     overview = "overview"
     clinical = "clinical"
-    radiological = "radiological"
+    radiologic = "radiologic"
     safety = "safety"
     other = "other"
 
+class Section(Enum):
+    abstract = "Abstract"
+    introduction = "Introduction"
+    methods = "Material and Methods"
+    results = "Results"
+    discussion = "Discussion"
+    conclusion = "Conclusion"
+    references = "References"
+    ifu = "IFU"
+
 class Parameter(BaseModel):
     name: str
     value: str
@@ -45,10 +65,9 @@ class ClinicalStudy(Article):
     tables: Optional[List[Table]] = Field(None, alias="tables")
     acceptance_year: Optional[int] = Field(None, alias="acceptance_year")
     acceptance_month: Optional[int] = Field(None, alias="acceptance_month")
-    sections: Optional[Dict[str]] = Field(None, alias="sections")
-    sectionCollection: Optional[Dict[str]] = Field(None, alias="sectionCollection")
+    sections: Optional[Dict[Section,str]] = Field(None, alias="sections")
 
-class DeviceIMF(Article):
+class DeviceIFU(Article):
     tables: Optional[List[Table]] = Field(None, alias="tables")
     sections: Optional[List[str]] = Field(None, alias="sections")
     sectionCollection: Optional[List[str]] = Field(None, alias="sectionCollection")

instruction.py

@@ -1,26 +1,82 @@
-
-
+from article import Assessment, Region, Section
 from pydantic import BaseModel, Field
 from typing import Optional, List, Dict
 from enum import Enum
 
-class Term(BaseModel):
-    id: int
-    terms: List[str] = Field(None, alias="terms")
-
-class PostParser(BaseModel):
-    id: int
-    name: str
-    terms: List[str] = Field(None, alias="terms")
-                             
-class PostParser(BaseModel):
-    id: int
-    name: str
-    terms: List[str] = Field(None, alias="terms")
-
-class Instruction(BaseModel):
-    name: str
-
-class InstructionClassifier(BaseModel):
-    terms: List[Term] = Field(None, alias="terms")
-    instruction: Instruction = Field(None, alias="instruction")
+from langchain.prompts.chat import ChatPromptTemplate
+from langchain.chat_models import ChatOpenAI
+from langchain.schema.runnable.base import RunnableSequence
+
+value_map = {
+    "overview":     Assessment.overview,
+    "clinical":     Assessment.clinical,
+    "radiologic":   Assessment.radiologic,
+    "safety":       Assessment.safety,
+    "other":        Assessment.other,
+    "spine":        Region.spine,
+    "extremity":    Region.extremity,
+    "all":          Region.all,
+    "abstract":     Section.abstract,
+    "introduction": Section.introduction,
+    "material and methods":      Section.methods,
+    "results":      Section.results,
+    "discussion":   Section.discussion,
+    "conclusion":   Section.conclusion,
+    "references":   Section.references,
+}
+
+class Parser(BaseModel):
+    term: str = Field("{term}", description="the pattern to be replaced with the output_term.")
+    region: Region = Field("cervical", alias="region")
+    assessment: Assessment = Field(None, alias="assessment")
+    replacement: str = Field(None, description="the term from last the input to be replaced with.")
+    
+    def parse(self,content):
+        content.replace(self.term,self.replacement)
+        return content
+
+# class Path(BaseModel):  # maybe too early to generalize this. Lets walk through a normal one for the insturction classifier first.
+#     name: str
+#     inputs: List[str] | str = Field([""], alias="inputs")
+#     variables: Dict[str,str] = Field({"term":""}, alias="variables")
+#     assessment: Assessment = Field(None, description="The clinical trail assessment steps")
+#     chain: RunnableSequence = Field([""], description="The nodes in the path to be executed.")
+    
+#     def run(self,article):
+#         content = " ".join([article[s] for s in self.inputs])
+#         self.varialbes.update(content=content)
+#         article[self.name] = self.chain.invoke(self.variables)
+
+#     async def arun(self,article):
+#         pass
+
+class ChainClassifier(BaseModel):
+    terms: List[str] = Field([""], alias="terms")
+    region: Region = Field(Region.spine, alias="region")
+    sections: Section = Field(None, alias="sections")
+    path: List[object] = Field(..., description="the automation path to be executed.")
+    chain: RunnableSequence = Field(None, description="the nodes in the path to be executed.")
+
+    def classify(self,article):
+        content = "".join([article[s] for s in self.sections])
+        if not self.validate_region(article):
+            return
+        if all([t in content for t in self.terms]):
+            return self.instruction
+    
+    def validate_region(self,article):
+        if self.region == Region.all:
+            return True
+        else:
+            return  self.region == article.region
+        
+    def parse_obj_mapped(self,obj,key_map):
+        for k,v in key_map.items():
+            if k in obj:
+                obj[v] = obj.pop(k)
+        self.parse_obj(obj)
+
+    def run(self,article):
+        self.chain.invoke(article)
+
+