src/agora/logging.py

+#!/usr/bin/env jupyter
+"""
+Add general logging functions and decorators
+"""
+
+import logging
+from time import perf_counter
+
+
+def timer(func):
+    # Log duration of a function into aliby logfile
+    def wrap_func(*args, **kwargs):
+        t1 = perf_counter()
+        result = func(*args, **kwargs)
+        logging.getLogger("aliby").debug(
+            f"{func.__qualname__} took {(perf_counter()-t1):.4f}s"
+        )
+        return result
+
+    return wrap_func

src/agora/utils/association.py

+#!/usr/bin/env jupyter

src/agora/utils/cast.py

+#!/usr/bin/env jupyter
+
+"""
+Convert some types to others
+"""
+
+
+def _str_to_int(x: str or None):
+    """
+    Cast string as int if possible. If Nonetype return None.
+    """
+    if x is not None:
+        try:
+            return int(x)
+        except:
+            return x

src/agora/utils/lineage.py

+#!/usr/bin/env python3
+import re
+import typing as t
+
+import numpy as np
+import pandas as pd
+
+from agora.io.bridge import groupsort
+from itertools import groupby
+
+
+def mb_array_to_dict(mb_array: np.ndarray):
+    """
+    Convert a lineage ndarray (trap, mother_id, daughter_id)
+    into a dictionary of lists ( mother_id ->[daughters_ids] )
+    """
+    return {
+        (trap, mo): [(trap, d[0]) for d in daughters]
+        for trap, mo_da in groupsort(mb_array).items()
+        for mo, daughters in groupsort(mo_da).items()
+    }
+

src/aliby/bin/__init__.py

+"""
+Command Line Interface utilities.
+"""

src/aliby/bin/run.py

+#!/usr/bin/env jupyter
+import argparse
+
+from agora.utils.cast import _str_to_int
+from aliby.pipeline import Pipeline, PipelineParameters
+
+
+def run():
+    """
+    Run a default microscopy analysis pipeline.
+
+    Parse command-line arguments and set default parameter values for running a pipeline, then
+    construct and execute the pipeline with the parameters obtained. Command-line arguments can
+    override default parameter values. If a command-line argument is a string representation of
+    an integer, convert it to an integer.
+
+    Returns
+    -------
+    None
+
+    Examples
+    --------
+    FIXME: Add docs.
+    """
+    parser = argparse.ArgumentParser(
+        prog="aliby-run",
+        description="Run a default microscopy analysis pipeline",
+    )
+
+    param_values = {
+        "expt_id": None,
+        "distributed": 2,
+        "tps": 2,
+        "directory": "./data",
+        "filter": 0,
+        "host": None,
+        "username": None,
+        "password": None,
+    }
+
+    for k in param_values:
+        parser.add_argument(f"--{k}", action="store")
+
+    args = parser.parse_args()
+
+    for k in param_values:
+        if passed_value := _str_to_int(getattr(args, k)):
+
+            param_values[k] = passed_value
+
+    params = PipelineParameters.default(general=param_values)
+    p = Pipeline(params)
+
+    p.run()