Writing all function signatures in Python

CMakeLists.txt

@@ -16,3 +16,5 @@ daq_add_application(create_config_plot create_config_plot.cxx GraphBuilder.cpp L
 daq_add_unit_test(Graph_test)
 
 daq_install()
+
+install(DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/python/ DESTINATION ${CMAKE_INSTALL_PYTHONDIR} OPTIONAL FILES_MATCHING PATTERN "py.typed" PATTERN "*.pyi")

python/daqconf/assets.py

@@ -1,29 +1,53 @@
 
-from os.path import exists,abspath,dirname,expandvars
+from os.path import exists, abspath, dirname, expandvars
+from urllib.parse import urlparse, parse_qsl
 
 from daq_assettools.asset_file import AssetFile
 from daq_assettools.asset_database import Database
 from sqlite3 import OperationalError
 
-def resolve_asset_file(data_file, verbose = False):
-    from urllib.parse import urlparse, parse_qsl
+
+def resolve_asset_file(data_file: str, verbose: bool = False) -> str:
+    """
+    Resolves a data file URI to an absolute file path. The data file can be specified as
+        - An asset URI (e.g., asset://?name=frames)
+        - A file URI (e.g., file:///path/to/frames.bin)
+        - A local file path (e.g., /path/to/frames.bin)
+
+    Args:
+        data_file (str): The data file URI or path to resolve.
+        verbose (bool): If True, prints additional information during resolution.
+
+    Returns:
+        str: The absolute path to the resolved data file.
+
+    Raises:
+        RuntimeError: If the data file cannot be found or resolved.
+    """
     data_file_url = urlparse(data_file)
 
     if verbose:
         print(f"Checking asset URI {data_file_url}")
 
     if data_file_url.scheme == 'asset':
         asset_query = dict(parse_qsl(data_file_url.query))
-        asset_db = Database('/cvmfs/dunedaq.opensciencegrid.org/assets/dunedaq-asset-db.sqlite')
+        asset_db = Database(
+            '/cvmfs/dunedaq.opensciencegrid.org/assets/dunedaq-asset-db.sqlite'
+        )
         asset_query['status'] = 'valid'
 
         try:
             files = asset_db.get_files(asset_query)
             if not files:
-                raise RuntimeError(f"Couldn\'t find a valid asset for the query {data_file_url.query}")
+                raise RuntimeError(
+                    f"Couldn\'t find a valid asset for the query {data_file_url.query}"
+                )
 
             elif len(files)>1:
-                print(f"Found {len(files)} assets in {dirname(asset_db.database_file)}, taking the first one")
+                print(
+                    f"Found {len(files)} assets in {dirname(asset_db.database_file)}, "
+                    "taking the first one"
+                )
 
             if verbose:
                 print(f"Found asset in {dirname(asset_db.database_file)}")

python/daqconf/cider/data_structures/configuration_handler.py

@@ -1,5 +1,5 @@
 import os
-from typing import Any, Dict, List
+from typing import Dict, List
 
 import conffwk 
 
@@ -43,11 +43,11 @@ def __cache_all_conf_objects(self)->None:
                 self._loaded_dals.append(conf_obj)
 
     #==============================  Getters + Setters ==============================#
-    def get_relationships_for_conf_object(self, conf_object)->List[Any]:
+    def get_relationships_for_conf_object(self, conf_object: object) -> List[Dict[str, object]]:
         """For a given configuration object, return all related objects
 
         Arguments:
-            conf_object -- Any DAL object
+            conf_object -- configuration DAL object
 
         Returns:
             List of related objects
@@ -67,7 +67,7 @@ def get_relationships_for_conf_object(self, conf_object)->List[Any]:
 
         return relations_list
 
-    def get_conf_objects_class(self, conf_class: str):
+    def get_conf_objects_class(self, conf_class: str) -> List[object]:
         """Get all configuration objects of a given class
 
         Arguments:
@@ -78,7 +78,7 @@ def get_conf_objects_class(self, conf_class: str):
         """        
         return self._configuration.get_dals(conf_class)
 
-    def get_all_conf_classes(self)->Dict[str, Any]:
+    def get_all_conf_classes(self) -> Dict[str, List[object]]:
         """Gets all classes + objects of that class in the configuration
 
         Returns:
@@ -122,7 +122,7 @@ def conf_obj_list(self):
         """        
         return self._loaded_dals
 
-    def get_obj(self, class_id: str, uid: str):
+    def get_obj(self, class_id: str, uid: str) -> object:
         """Get a particular configuration object 
 
         Arguments:
@@ -134,7 +134,7 @@ def get_obj(self, class_id: str, uid: str):
         """        
         return self.configuration.get_obj(class_id, uid)
 
-    def commit(self, update_message: str):
+    def commit(self, update_message: str) -> None:
         """Commit changes to the database
 
         Arguments:
@@ -149,7 +149,7 @@ def n_dals(self)->int:
         """        
         return len(self._loaded_dals)
 
-    def add_new_conf_obj(self, class_id: str, uid: str):
+    def add_new_conf_obj(self, class_id: str, uid: str) -> None:
         """Add new configuration object
 
     Arguments:
@@ -161,7 +161,7 @@ def add_new_conf_obj(self, class_id: str, uid: str):
         self.configuration.update_dal(config_as_dal)
         self._loaded_dals.append(config_as_dal)
 
-    def destroy_conf_obj(self, class_id: str, uid: str):
+    def destroy_conf_obj(self, class_id: str, uid: str) -> None:
         """Destroy a configuration object
 
         Arguments:
@@ -172,8 +172,14 @@ def destroy_conf_obj(self, class_id: str, uid: str):
         self.configuration.destroy_dal(dal)
         self._loaded_dals.remove(dal)
 
-    def modify_relationship(self, class_id, uid, relationship_name: str, updated_value,
-                            append: bool=False):
+    def modify_relationship(
+        self,
+        class_id: str,
+        uid: str,
+        relationship_name: str,
+        updated_value: object,
+        append: bool = False,
+    ) -> None:
         """Modify TODO: EDIT THIS
 
         :param class_id: _description_

python/daqconf/cider/widgets/modify_config_relations.py

@@ -8,15 +8,19 @@
 from textual.containers import VerticalScroll
 from rich.console import RichCast, ConsoleRenderable
 
-from typing import Any
+from typing import Protocol
 
 from daqconf.cider.widgets.configuration_controller import ConfigurationController
 from daqconf.cider.widgets.custom_rich_log import RichLogWError
 
 
 
+class RelatedDal(Protocol):
+    def __repr__(self) -> str: ...
+
+
 class SingleRelationshipModifier(Static):
-    def __init__(self, relationship_type: str, current_related_dal: Any, relationship_name: str,
+    def __init__(self, relationship_type: str, current_related_dal: RelatedDal | object, relationship_name: str,
                  renderable: ConsoleRenderable | RichCast | str = "", *, expand: bool = False,
                  shrink: bool = False, markup: bool = True, name: str | None = None, id: str | None = None,
                  classes: str | None = None, disabled: bool = False) -> None:

python/daqconf/cider/widgets/popups/edit_cell_screen.py

@@ -5,17 +5,21 @@
 from textual.widgets import Input, Label
 from textual.containers import Container
 
-from typing import Any
+from typing import Protocol
 
 from daqconf.cider.widgets.configuration_controller import ConfigurationController
 
+class TableCellEvent(Protocol):
+    row_key: object
+
+
 class EditCellScreen(ModalScreen):
     css_file_path = f"{environ.get('DAQCONF_SHARE')}/config/textual_dbe/textual_css"
 
     CSS_PATH = f"{css_file_path}/edit_cell_layout.tcss"
 
     def __init__(
-        self, event: Any, name: str | None = None, id: str | None = None, classes: str | None = None) -> None:
+        self, event: TableCellEvent, name: str | None = None, id: str | None = None, classes: str | None = None) -> None:
         super().__init__(name=name, id=id, classes=classes)
         """Screen which pops up when a cell is clicked in the ConfigTable
         """

python/daqconf/cider/widgets/selection_menu.py

@@ -1,4 +1,3 @@
-from typing import Any
 import numpy as np
 
 from textual.widgets import Static, Tree

python/daqconf/consolidate.py

@@ -7,7 +7,20 @@
 log = getLogger('daqconf.consolidate')
 
 
-def get_all_includes(db, file):
+def get_all_includes(db: conffwk.Configuration, file: str) -> list[str]:
+    """
+    Recursively get all includes from an OKS database
+
+    Args:
+        db (conffwk.Configuration): The OKS database to get includes from
+        file (str): The file to get includes from
+
+    Returns:
+        list(str): A list of all includes in the database
+
+    Raises:
+        None
+    """
     includes = db.get_includes(file)
     for include in includes:
         if "data.xml" not in include:
@@ -16,12 +29,26 @@ def get_all_includes(db, file):
 
     return list(set(includes))
 
-def consolidate_db(oksfile: str, output_file: str, session_id: Optional[str] = None)->None:
-    """Consolidates a single session
+def consolidate_db(
+    oksfile: str, output_file: str, session_id: Optional[str] = None
+) -> None:
+    """
+    Consolidates a session database files into a single file.
+
+    The consolidated file contains all the schema and data required to define the entire
+    session. If session_id is None, all sessions will be consolidated into the single
+    output file.
+
+    Args:
+        oksfile: OKS file(s) to consolidate
+        output_file: File to output consolidated database to 
+        session_id: Name of session, defaults to None
+
+    Returns:
+        None
 
-    :param oksfile: OKS file(s) to consolidate
-    :param output_file: File to output consolidated database to 
-    :param session_id: Name of session, defaults to None
+    Raises:
+        None
     """    
     log.info(f"Consolidating database into output database '{output_file}'. Input database: '{oksfile}'.")
 
@@ -38,13 +65,16 @@ def consolidate_db(oksfile: str, output_file: str, session_id: Optional[str] = N
 
 
 def create_copy_template(oksfile: str, output_file: str)->Tuple[conffwk.Configuration, conffwk.Configuration]:
-    '''
-    Creates a blank oks .data.xml file stored in output_file with all the schema includes of oksfile
-    :param oksfile: OKS file to copy includes from
-    :param output_file: OKS file to copy includes into
+    """
+    Create a blank oks .data.xml template file with all the schema includes of oksfile.
 
-    :returns: Tuple of old_db, copied_db
-    '''
+    Args:
+        oksfile: OKS file to copy includes from
+        output_file: OKS file to copy includes into
+
+    Returns:
+        Tuple of old_db, copied_db
+    """
     log.debug("Reading database")
     db = conffwk.Configuration("oksconflibs:" + oksfile)
 
@@ -89,7 +119,7 @@ def consolidate_session(db: conffwk.Configuration, new_db: conffwk.Configuration
     dal_list = get_relationships(db, dal_session, [])
     copy_dals_to_cfg(new_db, dal_list)
 
-def get_relationships(db: conffwk.Configuration, current_dal, dal_list):
+def get_relationships(db: conffwk.Configuration, current_dal: conffwk.Configuration, dal_list: list[conffwk.Configuration]) -> list[conffwk.Configuration]:
     '''
     Recurssively get all objects related to current_dal
     '''
@@ -109,7 +139,7 @@ def get_relationships(db: conffwk.Configuration, current_dal, dal_list):
     return dal_list
 
 
-def copy_dals_to_cfg(new_db: conffwk.Configuration, dal_list)->None:
+def copy_dals_to_cfg(new_db: conffwk.Configuration, dal_list: list[conffwk.Configuration]) -> None:
     '''
     Copy a list of dals into a configuration
     '''
@@ -121,7 +151,7 @@ def copy_dals_to_cfg(new_db: conffwk.Configuration, dal_list)->None:
     new_db.commit()
 
 
-def copy_configuration(dest_dir : Path, input_files: list):
+def copy_configuration(dest_dir : Path, input_files: list[conffwk.Configuration]) -> list[conffwk.Configuration]:
     if len(input_files) == 0:
         return []
 
@@ -157,7 +187,7 @@ def copy_configuration(dest_dir : Path, input_files: list):
     return output_dbs
 
 
-def consolidate_files(oksfile, *input_files):
+def consolidate_files(oksfile: str, *input_files) -> None:
     includes = []
     dbs = []
     str_in_files = '\n'.join(input_files)

python/daqconf/createOKSdb.py

@@ -3,7 +3,7 @@
 import glob
 
 
-def generate_file(oksfile, include):
+def generate_file(oksfile: str, include: list[str]) -> None:
     """Simple script to create an 'empty' OKS file.
     The file will automatically include the confmodel schema
     and any other OKS files you specify"""

python/daqconf/dal_helpers.py

@@ -3,25 +3,25 @@
 ##
 # Dal helpers
 #
-def get_attribute_info(o):
+def get_attribute_info(o: object) -> dict[str, object]:
     return o.__schema__['attribute']
 
-def get_relation_info(o):
+def get_relation_info(o: object) -> dict[str, object]:
     return o.__schema__['relation']
 
-def get_attribute_list(o):
+def get_attribute_list(o: object) -> list[str]:
     return list(get_attribute_info(o))
 
-def get_relation_list(o):
+def get_relation_list(o: object) -> list[str]:
     return list(get_relation_info(o))
 
-def get_superclass_list(o):
+def get_superclass_list(o: object) -> list[str]:
     return o.__schema__['superclass']
 
-def get_subclass_list(o):
+def get_subclass_list(o: object) -> list[str]:
     return o.__schema__['subclass']
 
-def compare_dal_obj(a, b):
+def compare_dal_obj(a: object, b: object) -> bool:
     """Compare two dal objects by content"""
 
     # TODO: add a check on a and b being dal objects
@@ -45,7 +45,7 @@ def compare_dal_obj(a, b):
 
 
 #---------------
-def find_related(dal_obj, dal_group: set):
+def find_related(dal_obj: object, dal_group: set[object]) -> None:
 
 
     rels = get_relation_list(dal_obj)
@@ -71,7 +71,7 @@ def find_related(dal_obj, dal_group: set):
         find_related(o, dal_group)
 
 from collections.abc import Iterable
-def find_duplicates( collection: Iterable ):
+def find_duplicates(collection: Iterable[object]) -> set[object]:
     """
     Find duplicated dal objects in a collection by comparing objects attributes and relationships
     """

python/daqconf/dromap2oks.py

@@ -3,7 +3,13 @@
 import json
 import sys
 
-def dro_json_to_oks(jsonfile, oksfile, source_id_offset, nomap, lcores):
+def dro_json_to_oks(
+    jsonfile: str,
+    oksfile: str,
+    source_id_offset: int,
+    nomap: bool,
+    lcores: str,
+) -> None:
     """Simple script to convert a JSON readout map file to an OKS file."""
 
     group_name = os.path.basename(jsonfile).removesuffix(".json")