Merge pull request #25 from wwPDB/DAOTHER-9213

PCM/PTM tab in AnnMod

requirements.txt

@@ -12,5 +12,6 @@ wwpdb.utils.detach
 mmcif.utils
 matplotlib
 pygal
+pytest
 
 mock; python_version < '3'

tox.ini

@@ -47,6 +47,9 @@ minversion = 3.4.0
 skip_missing_interpreters = true
 skipsdist = false
 
+[pytest]
+python_files = "*PyTest.py"
+
 [testenv]
 passenv = CONFIG_SUPPORT_TOKEN_ENV,OE_LICENSE
 allowlist_externals = echo
@@ -77,6 +80,7 @@ commands =
     {envpython} -m unittest discover -v --start-directory {[local_settings]test_path_2} --pattern "{[local_settings]test_pattern}"
     {envpython} -m unittest discover -v --start-directory {[local_settings]test_path_3} --pattern "{[local_settings]test_pattern}"
     {envpython} -m unittest discover -v --start-directory {[local_settings]test_path_4} --pattern "{[local_settings]test_pattern}"
+    {envpython} -m pytest -v {[local_settings]test_path_1} --tb=short
     echo "Completed {envname}"
 
 #

wwpdb/apps/ann_tasks_v2/pcm/PcmCCDEditorForm.py

@@ -0,0 +1,163 @@
+"""
+Manage the generating of PCM missing annotation table.
+
+"""
+
+try:
+    import cPickle as pickle
+except ImportError:
+    import pickle as pickle
+
+import os
+import csv
+import sys
+import logging
+import traceback
+
+from wwpdb.utils.dp.RcsbDpUtility import RcsbDpUtility
+import logging
+
+# Set the default logger handler level to INFO
+logging.getLogger().setLevel(logging.INFO)
+
+
+class PcmCCDEditorForm(object):
+    """
+    The CoordEditorForm class generates coordinate editor form.
+
+    """
+
+    def __init__(self, reqObj=None, verbose=False, log=sys.stderr):
+        self.__verbose = verbose
+        self.__lfh = log
+        self.__reqObj = reqObj
+        self.__entryFile = None
+        #
+        self.__setup()
+
+    def __setup(self):
+        self.__siteId = self.__reqObj.getValue("WWPDB_SITE_ID")
+        self.__sObj = self.__reqObj.getSessionObj()
+        self.__sessionPath = self.__sObj.getPath()
+        self.__entryId = self.__reqObj.getValue("entryid")
+        self.__identifier = self.__reqObj.getValue("display_identifier")
+        self.__csvPath = os.path.join(self.__sessionPath, self.__entryId + "_ccd_no_pcm_ann.csv")
+        self.__entryFile = None
+        #
+        self.__listItemTemplate = (
+            '<li><a id="%s" class="discontrol ui-corner-all" href="#">'
+            + '<span class="ui-icon fltlft ui-icon-circle-arrow-e"></span> %s </a>\n'
+            + '<div id="display_%s" style="display:none"></div>\n</li>\n'
+        )
+        #
+        self.__tableTemplate = '<table id="table_%s" class="table table-condensed table-bordered table-striped">\n'
+        self.__tdTagTemplate = '<td style="border-style:none">%s</td>\n'
+        self.__editableTemplate = '<b class="%s" id="%s" style="display:inline">%s</b>'
+
+    def setLogHandle(self, log=sys.stderr):
+        """Reset the stream for logging output."""
+        try:
+            self.__lfh = log
+            return True
+        except:  # noqa: E722 pylint: disable=bare-except
+            return False
+
+    def run(self):
+        """Generate JSON format data"""
+        if not self.__identifier:
+            return False
+        #
+        if self.__identifier == self.__entryId:
+            self.__runPcmCcdCheck()
+
+        if os.access(self.__csvPath, os.F_OK):
+            return True
+        #
+        return False
+
+    def __runPcmCcdCheck(self):
+        """Run X program"""
+        self.__entryFile = self.__reqObj.getValue("entryfilename")
+        entry_file_path = os.path.join(self.__sessionPath, self.__entryFile)
+
+        if not os.access(entry_file_path, os.F_OK):
+            logging.error("Missing entry file %s", entry_file_path)
+            return
+        #
+        logging.info("Processing entry file %s", self.__entryFile)
+
+        try:
+            dp = RcsbDpUtility(tmpPath=self.__sessionPath, siteId=self.__siteId, verbose=self.__verbose, log=self.__lfh)
+            dp.imp(entry_file_path)
+            dp.op("annot-pcm-check-ccd-ann")
+            dp.exp(dstPath=self.__csvPath)
+            dp.expLog(os.path.join(self.__sessionPath, self.__entryId + "_ccd_no_pcm_ann.log"))
+            dp.cleanup()
+        except:  # noqa: E722 pylint: disable=bare-except
+            traceback.print_exc(file=self.__lfh)
+        #
+
+    def getCCDForm(self):
+        """Get CCD missing annotation form"""
+        myD = {}
+        myD["statuscode"] = "failed"
+        myD["statustext"] = "Invalid ajax call"
+        if not self.__identifier:
+            return myD
+        #
+        if self.__identifier == self.__entryId:
+            return self.__buildCCDFormHtml()
+        #
+        return myD
+
+    def __buildCCDFormHtml(self):
+        # depending on the output of the binary
+        # check if it matches this entry id
+
+        if not os.access(self.__csvPath, os.F_OK):
+            myD = {}
+            myD["statuscode"] = "failed"
+            myD["statustext"] = "Failed to build form for x"
+            return myD
+
+        htmlcontent = self.__tableTemplate % self.__identifier
+        htmlcontent += (
+            "<tr>\n<th>Comp Id</th>\n<th>Modified Residue Id</th>\n<th>Type</th>\n<th>Category</th>\n<th>Position</th>\n<th>Polypeptide Position</th>\n<th>Comp Id Linking Atom</th>\n<th>Modified Residue Id Linking Atom</th>\n<th>First Instance Model Db Code</th>\n<th>ChemRefUI Link</th>\n</tr>\n"
+        )
+        #
+        columns = {
+            "Comp_id": {"editable": False},
+            "Modified_residue_id": {"editable": False},
+            "Type": {"editable": False},
+            "Category": {"editable": False},
+            "Position": {"editable": False},
+            "Polypeptide_position": {"editable": False},
+            "Comp_id_linking_atom": {"editable": False},
+            "Modified_residue_id_linking_atom": {"editable": False},
+            "First_instance_model_db_code": {"editable": False},
+        }
+
+        with open(self.__csvPath, 'r') as csvfile:
+            reader = csv.DictReader(csvfile)
+            for row in reader:
+                comp_id = row['Comp_id']
+
+                for c in columns:
+                    if columns[c]['editable']:
+                        ed_id = f"{comp_id}_{c}"
+                        ed_text = self.__editableTemplate % ("editable_text", ed_id, row[c])
+                        htmlcontent += self.__tdTagTemplate % ed_text
+                    else:
+                        htmlcontent += self.__tdTagTemplate % row[c]
+
+                # add last column with link
+                chem_ref_link = "/chem_ref_data_ui/chemref_search_ref_bs.html?searchTarget=%s" % comp_id
+                htmlcontent += self.__tdTagTemplate % f'<a href="{chem_ref_link}" target="_blank">{comp_id}</a>'
+                htmlcontent += "</tr>\n"        
+        #
+        htmlcontent += "</table>\n"
+        #
+        myD = {}
+        myD["statuscode"] = "ok"
+        myD["htmlcontent"] = htmlcontent
+        return myD

wwpdb/apps/ann_tasks_v2/webapp/AnnTasksWebAppWorker.py

@@ -119,6 +119,7 @@ def __init__(self, reqObj=None, verbose=False, log=sys.stderr):
             "/service/ann_tasks_v2/nafeaturescalc": "_naFeaturesCalcOp",
             "/service/ann_tasks_v2/secstructcalc": "_secondaryStructureCalcOp",
             "/service/ann_tasks_v2/transformcoordcalc": "_transformCoordCalcOp",
+            "/service/ann_tasks_v2/pcm_get_ccd_form": "_pcmGetCCDFormOp",
             #
             "/service/ann_tasks_v2/mergexyzcalc": "_mergeXyzCalcOp",
             "/service/ann_tasks_v2/terminalatomscalc": "_terminalAtomsCalcOp",

wwpdb/apps/ann_tasks_v2/webapp/CommonTasksWebAppWorker.py

@@ -94,6 +94,9 @@
 from wwpdb.apps.ann_tasks_v2.editCoord.CSEditorForm import CSEditorForm
 from wwpdb.apps.ann_tasks_v2.editCoord.CSEditorUpdate import CSEditorUpdate
 
+#
+from wwpdb.apps.ann_tasks_v2.pcm.PcmCCDEditorForm import PcmCCDEditorForm
+
 #
 from wwpdb.apps.ann_tasks_v2.editCoord.CoordEditorForm_v2 import CoordEditorForm
 from wwpdb.apps.ann_tasks_v2.editCoord.CoordEditorUpdate import CoordEditorUpdate
@@ -1132,6 +1135,42 @@ def _transformCoordCalcOp(self):
 
         return rC
 
+    def _pcmGetCCDFormOp(self):
+        """Generating CCD missing annotation form"""
+        if self._verbose:
+            self._lfh.write("+CommonTasksWebAppWorker._pcmGetCCDFormOp() starting\n")
+        self._getSession(useContext=True)
+        #
+        dU = DetachUtils(reqObj=self._reqObj, verbose=self._verbose, log=self._lfh)
+        ccdEditorFormOp = PcmCCDEditorForm(reqObj=self._reqObj, verbose=self._verbose, log=self._lfh)
+        #
+        myD = {}
+        sph = self._reqObj.getSemaphore()
+        delayValue = self._reqObj.getValue("delay")
+        if sph:
+            if dU.semaphoreExists(sph):
+                myD = ccdEditorFormOp.getCCDForm()
+            else:
+                time.sleep(int(delayValue))
+                myD["statuscode"] = "running"
+            #
+        else:
+            entryId = self._reqObj.getValue("entryid")
+            identifier = self._reqObj.getValue("display_identifier")
+            dU.set(workerObj=ccdEditorFormOp, workerMethod="run")
+            dU.runDetach()
+            myD["statuscode"] = "running"
+
+            # if (identifier == entryId) or identifier.startswith("chain_"):
+            # else:
+            #     myD = ccdEditorFormOp.get()
+            #
+        #
+        rC = ResponseContent(reqObj=self._reqObj, verbose=self._verbose, log=self._lfh)
+        rC.setReturnFormat("json")
+        rC.addDictionaryItems(myD)
+        return rC
+
     def _mergeXyzCalcOp(self):
         """ """
         if self._verbose:

wwpdb/apps/tests-ann/PcmPyTest.py

@@ -0,0 +1,81 @@
+import os
+import pytest
+import tempfile
+
+from wwpdb.utils.config.ConfigInfoData import ConfigInfoData
+from wwpdb.apps.ann_tasks_v2.pcm.PcmCCDEditorForm import PcmCCDEditorForm
+
+
+MOCK_SESSIONS_PATH = tempfile.mkdtemp()
+
+@pytest.fixture
+def mock_config(monkeypatch):
+    mc = {
+        "SITE_REFDATA_PROJ_NAME_PRDCC": "",
+        "SITE_WEB_APPS_SESSIONS_PATH": MOCK_SESSIONS_PATH,
+        "SITE_WEB_APPS_TOP_SESSIONS_PATH": MOCK_SESSIONS_PATH,
+        "SITE_REFDATA_PROJ_NAME_CC": "",
+        "SITE_CC_DICT_PATH": "",
+        "SITE_REFDATA_TOP_CVS_SB_PATH": "",
+        "PDBX_DICTIONARY_NAME_DICT": {"DEPOSIT": ""},
+        "REFERENCE_PATH": "",
+        "SITE_PDBX_DICT_PATH": "",
+        "SITE_PACKAGES_PATH": "",
+        "SITE_LOCAL_APPS_PATH": "",
+    }
+
+    monkeypatch.setattr(ConfigInfoData, "getConfigDictionary", lambda s: mc)
+
+
+class MockReqObj:
+    def __init__(self, values):
+        self.values = values
+        self.session = MockSessionObj()
+
+    def getValue(self, key):
+        return self.values.get(key)
+
+    def getSessionObj(self):
+        # Return a mock session object
+        return self.session
+
+
+class MockSessionObj:
+    def __init__(self) -> None:
+        self._session = os.path.join(MOCK_SESSIONS_PATH, "mock_session")
+        os.makedirs(self._session, exist_ok=True)
+
+    def getPath(self):
+        return self._session
+
+
+def test_run_binary(mock_config):
+    mock_req = MockReqObj({"display_identifier": "1cbs", "entryid": "1cbs", "entryfilename": "1cbs.cif", "WWPDB_SITE_ID": "WWPDB"})
+    pcm_form = PcmCCDEditorForm(reqObj=mock_req, verbose=False)
+    open(os.path.join(mock_req.session.getPath(), "1cbs.cif"), "w").close()
+
+    assert pcm_form.run() == True
+    assert os.path.exists(os.path.join(mock_req.session.getPath(), "1cbs_ccd_no_pcm_ann.csv"))
+    assert os.path.exists(os.path.join(mock_req.session.getPath(), "1cbs_ccd_no_pcm_ann.log"))
+
+
+def test_get_html_form(mock_config):
+    mock_req = MockReqObj({"display_identifier": "1cbs", "entryid": "1cbs", "entryfilename": "1cbs.cif", "WWPDB_SITE_ID": "WWPDB"})
+    pcm_form = PcmCCDEditorForm(reqObj=mock_req, verbose=False)
+
+    with open(os.path.join(mock_req.session.getPath(), "1cbs_ccd_no_pcm_ann.csv"), "w") as fp:
+        fp.write("Comp_id,Modified_residue_id,Type,Category,Position,Polypeptide_position,Comp_id_linking_atom,Modified_residue_id_linking_atom,First_instance_model_db_code\n")
+        fp.write("MLU,missing,missing,missing,missing,missing,.,.,1PN3\n")
+        fp.write("OMZ,missing,missing,missing,missing,missing,.,.,1PN3\n")
+        fp.write("GHP,missing,missing,missing,missing,missing,.,.,1PN3\n")
+        fp.write("BGC,GHP,missing,missing,missing,missing,C1,O4,1PN3\n")
+        fp.write("OMY,missing,missing,missing,missing,missing,.,.,1PN3\n")
+        fp.write("3FG,missing,missing,missing,missing,missing,.,.,1PN3")
+
+    html_form = pcm_form.getCCDForm()["htmlcontent"]
+    assert "MLU" in html_form
+    assert "OMZ" in html_form
+    assert "GHP" in html_form
+    assert "BGC" in html_form
+    assert "OMY" in html_form
+    assert "3FG" in html_form