added the option to create an csh session

app/metrics/assessments_lifespan.py

@@ -20,10 +20,11 @@ async def get_tasks_definitions(app: FastAPI):
     :return: None
     """
     regex = re.compile(r"^CA-RDA-([FAIR][1-9](\.[0-9])?)-")
+    regex_csh = re.compile(r"^CSH-RDA-([FAIR][1-9](\.[0-9])?)-")
     config = get_settings()
 
     def parse_line(line):
-        sub_group = regex.search(line["TaskName"]).groups()[0]
+        sub_group = regex_csh.search(line["TaskName"]).groups()[0]
         task_group = sub_group[0]
         return {
             line["TaskName"]: models.Indicator(
@@ -38,7 +39,7 @@ def parse_line(line):
         }
 
     # Get the list of tasks and their definitions from internal file
-    with open(config.indicators_path, "r") as file_handler:
+    with open(config.csh_indicators_path, "r") as file_handler:
         csv_reader = DictReader(file_handler, dialect="unix")
         [fair_indicators.update(parse_line(line)) for line in csv_reader]
 

app/models/csh.py

@@ -9,8 +9,8 @@ class Score(BaseModel):
     R: int = 0
 
 class study_evaluation: 
-    def __init__(self,json_data, schema):
-        print("TEWESTTS")
+    def __init__(self, json_data, schema):
+        print("+++++++++++ INSIDE study_evaluation")
         self.schema = schema
         self.metadata = json_data
         self.score = Score()

app/models/session.py

@@ -46,11 +46,13 @@ class SubjectType(str, Enum):
     - *url*: The archive/model to evaluate is at a specific url
     - *file*: The archive/model file is directly provided by the user
     - *manual*: No file is provided, the user will assess themselves the archive/model
+    - *csh* A JSON is provided, containing metadata from a Central Study Hub 
     """
 
     url = "url"
     file = "file"
     manual = "manual"
+    csh = "csh"
 
 
 @as_form
@@ -74,6 +76,7 @@ class SessionSubjectIn(BaseModel):
     - *is_biomodel*: Whether the model comes from BioModel (required attribute if *subject_type* is **manual**)
     - *is_pmr*: Whether the model comes from PMR (required attribute if *subject_type* is **manual**)
     - *subject_type*: See SubjectType model
+    - *metadata*: metadata from CSH
     """
 
     path: Union[HttpUrl, FileUrl, FilePath, None] = None
@@ -86,10 +89,14 @@ class SessionSubjectIn(BaseModel):
     is_archive_metadata_standard: Optional[bool]
     is_biomodel: Optional[bool]
     is_pmr: Optional[bool]
+    random: Optional[bool]
+    metadata: object
     subject_type: SubjectType
 
+
     @validator("subject_type", always=True)
     def necessary_data_provided(cls, subject_type: str, values: dict):
+        print(values)
         if subject_type is SubjectType.manual:
             if (
                 values.get("has_archive") is None
@@ -106,6 +113,9 @@ def necessary_data_provided(cls, subject_type: str, values: dict):
         elif subject_type is SubjectType.url:
             if values.get("path") is None:
                 raise ValueError("Url assessments need a url")
+        elif subject_type is SubjectType.csh:
+            if (values.get("metadata") is None):
+                raise ValueError("CSH assessments need a JSON object")
         return subject_type
 
     def dict(self, **kwargs):
@@ -192,7 +202,7 @@ def __init__(self, session: Session) -> None:
         self.assessed_data: Optional["CombineArchive"] = None
 
         if not session.tasks:
-            if self.user_input.subject_type is not SubjectType.manual:
+            if self.user_input.subject_type not in [SubjectType.manual, SubjectType.csh]:
                 self.assessed_data = self.retrieve_data(self.user_input.path)
             self.create_tasks()
 
@@ -259,6 +269,20 @@ def build_task(task_dict):
         session = Session(**session_json, session_subject=subject, tasks=tasks)
         return cls(session)
 
+    @classmethod
+    def from_csh(cls, session_id: str, session_data: SessionSubjectIn) -> "SessionHandler":
+        """
+        Creates a session based on A JSON from CSH
+
+        :param session_id: The session identifier that will be used
+        :param session_data:
+        :return: A SessionHandler object
+        """
+
+        session = Session(id=session_id, session_subject=session_data)
+        return cls(session)
+
+
     def _build_tasks_dict(self, tasks: list[Task]):
         """
         Creates the dictionary mapping all indicators names to their corresponding

app/routers/csh_router.py

@@ -41,4 +41,3 @@ def csh_study() -> Score: #metadata, schema_version
     print("?")
     return score
 
-print("Checki-di-check-check")

app/routers/router.py

@@ -22,11 +22,12 @@
 
 base_router = APIRouter()
 
-
+print("TESTSTSTS")
 @base_router.post("/session", tags=["Sessions"])
 def create_session(
     subject: SessionSubjectIn = Depends(SessionSubjectIn.as_form),
     uploaded_file: Optional[UploadFile] = None,
+    metadata: Optional[object] = None
 ) -> Session:
     """
     Create a new session based on user input
@@ -40,9 +41,14 @@ def create_session(
     \f
     :param subject: Pydantic model containing user input.
     :param uploaded_file: If subject type is 'file', this contains the uploaded omex archive.
+    :param metadata: If subject is "metadata" this object contains metadata from the CSH
     :return: The created session
     """
     session_id = str(uuid.uuid4())
+
+    print("created session " + session_id)
+
+
     if subject.subject_type is SubjectType.url:
         raise HTTPException(
             501, "The api only supports manual assessments at the moment"
@@ -63,8 +69,15 @@ def create_session(
             subject.path = path
         finally:
             uploaded_file.file.close()
+    elif subject.subject_type is SubjectType.csh:
+        print(subject)
+        print("!!!!!!!!!!!!!!!!!!!!!!!!!!!!! SUBJECT TYPE CSH DETECTED!")
+        if subject.metadata is None:
+            raise HTTPException(
+                422, "No JSON object was attached for assessment. Impossible to process query"
+            )
     try:
-        session_handler = SessionHandler.from_user_input(session_id, subject)
+        session_handler = SessionHandler.from_user_input(session_id, subject)  #is the only for form input? My impression: i
     except ValueError as e:
         raise HTTPException(422, str(e))
     try:

tests/factories.py

@@ -16,6 +16,7 @@
 
 
 class ManualSessionSubjectFactory(factory.Factory):
+    print("manual factory")
     class Meta:
         model = SessionSubjectIn
 
@@ -32,6 +33,7 @@ class Meta:
 
 
 class UrlSessionSubjectFactory(factory.Factory):
+    print("url factory")
     class Meta:
         model = SessionSubjectIn
 
@@ -40,6 +42,7 @@ class Meta:
 
 
 class FileSessionSubjectFactory(factory.Factory):
+    print("file factory")
     class Meta:
         model = SessionSubjectIn
 
@@ -55,6 +58,12 @@ def path(obj, create, extracted):
         logger.warning(f"Setting session path to {filename}")
         return filename
 
+class CshSessionSubjectFactory(factory.Factory):
+    print("csh factory")
+    class Meta:
+        model = SessionSubjectIn
+
+    subject_typ = "csh"
 
 class SessionFactory(factory.Factory):
     class Meta:

tests/local_request.py

@@ -0,0 +1,30 @@
+import requests
+import json
+
+print("TESTING THE SERVER")
+
+# Define the URL of the local server
+url = 'http://localhost:8000/session'
+
+# Define the data you want to send in the POST request (as a dictionary)
+metadata = {
+    "k1": "v1",
+    "k2": "v2"
+}
+
+body = {
+    'subject_type': 'csh',  # Make sure subject_type is provided
+    'metadata': json.dumps(metadata)
+}
+
+
+# Send an HTTP POST request with JSON content
+response = requests.post(url, data=body)
+
+# Check the response
+if response.status_code == 200:
+    print("Request was successful.")
+    print("Response:", response.json())
+else:
+    print("Request failed with status code:", response.status_code)
+    print(response.text)