discover_more_catalogs.py

"""
Identify new MongoDB catalogs for the tiled server.

This code prints additional lines that could be added
to the tiled `config.yml` file.  The lines describe catalogs
with known intake descriptions that are not already configured
in the `config.yml` file.
"""

import pathlib

import yaml


def tiled_test():
    from tiled.client import from_uri

    client = from_uri("http://otz:8000", "dask")
    cat = client["6idd"]
    print(f"{cat=}")


def main():
    home = pathlib.Path.home()
    # print(f"{home=}")
    databroker_configs = home / ".local" / "share" / "intake"
    # print(f"exists:{databroker_configs.exists()}  {databroker_configs}")

    master = {}
    for intake_yml in databroker_configs.iterdir():
        if intake_yml.is_file() and intake_yml.suffix == ".yml":
            # print(f"{item.suffix=}  {item=}")
            with open(intake_yml) as f:
                db_cfg = yaml.load(f, Loader=yaml.Loader)
                master.update(**db_cfg["sources"])
    # print(f"{len(master)}  {list(master.keys())}")

    local_config = pathlib.Path(__file__).parent / "config.yml"
    # print(f"exists:{local_config.exists()}  {local_config}")
    with open(local_config) as f:
        config = yaml.load(f, Loader=yaml.Loader)
        trees = {tree["path"]: tree for tree in config.get("trees", {})}

    new_entries = []
    for k, source in master.items():
        if k not in trees:
            if source.get("driver") == "bluesky-mongo-normalized-catalog":
                uri = source.get("args", {}).get("metadatastore_db")
                if uri is not None:
                    entry = dict(
                        path=k,
                        tree="databroker.mongo_normalized:Tree.from_uri",
                        args=dict(uri=uri),
                    )
                    new_entries.append(entry)
    print(yaml.dump(new_entries))


if __name__ == "__main__":
    main()