Template generation module for ENRGDAQ configurations.

Uses msgspec's built-in JSON schema generation to create schemas for: - Store configurations (CSV, ROOT, Redis, etc.) - DAQ job configurations (DAQJobTest, DAQJobHealthcheck, etc.) - DAQ job messages (DAQJobMessageStop, DAQJobMessageStore, etc.)

Usage

from enrgdaq.daq.template import ( get_store_config_templates, get_daq_job_config_templates, get_message_templates, )

Get store config templates (JSON schemas)

store_templates = get_store_config_templates()

Get DAQ job config templates (JSON schemas)

job_templates = get_daq_job_config_templates()

Get message templates (JSON schemas)

message_templates = get_message_templates()

Sub-modules

enrgdaq.daq.template.config

DAQ Job configuration template generator …

enrgdaq.daq.template.message

DAQ Job message template generator …

enrgdaq.daq.template.store

Store configuration template generator …

Functions

def get_daq_job_config_templates() ‑> dict[str, dict]

Expand source code

def get_daq_job_config_templates() -> dict[str, dict]:
    """
    Generate DAQ job configuration templates using msgspec's JSON schema generation.

    Returns a dictionary of JSON schemas for each DAQ job type.
    """
    templates: dict[str, dict] = {}

    for daq_job_class in get_all_daq_job_types():
        # Skip base DAQJob class
        if daq_job_class is DAQJob:
            continue

        # Get the config type for this job
        config_type = getattr(daq_job_class, "config_type", None)
        if config_type is None:
            continue

        job_name = daq_job_class.__name__

        # Generate JSON schema using msgspec
        # Some config types may have unsupported types, skip those
        try:
            schema = msgspec.json.schema(config_type)
        except TypeError:
            # Skip configs with unsupported types
            continue

        # Add our custom metadata
        schema["type_key"] = job_name
        schema["label"] = _get_job_label(job_name)
        # Use job class docstring for description (more relevant than config docstring)
        schema["description"] = _parse_class_docstring(daq_job_class)

        templates[job_name] = schema

    return templates

Generate DAQ job configuration templates using msgspec's JSON schema generation.

Returns a dictionary of JSON schemas for each DAQ job type.

def get_message_templates() ‑> dict[str, dict]

Expand source code

def get_message_templates() -> dict[str, dict]:
    """
    Generate DAQ job message templates using msgspec's JSON schema generation.

    Returns a dictionary of JSON schemas for each message type that can be
    sent to DAQ jobs via CNC.
    """
    # List of message types that can be sent to DAQ jobs
    # These are the types that the send_message handler can decode and route
    templates: dict[str, dict] = {}

    for message_cls in all_subclasses(DAQJobMessage):
        message_name = message_cls.__name__

        # Generate JSON schema using msgspec
        try:
            schema = msgspec.json.schema(message_cls)
        except TypeError:
            # Skip messages with unsupported types
            continue

        # Add our custom metadata
        schema["type_key"] = message_name
        schema["label"] = _get_message_label(message_name)
        schema["description"] = _parse_class_docstring(message_cls)

        templates[message_name] = schema

    return templates

Generate DAQ job message templates using msgspec's JSON schema generation.

Returns a dictionary of JSON schemas for each message type that can be sent to DAQ jobs via CNC.

def get_store_config_templates() ‑> dict[str, dict]

Expand source code

def get_store_config_templates() -> dict[str, dict]:
    """
    Generate store configuration templates using msgspec's JSON schema generation.

    Returns a dictionary of JSON schemas for each store type.
    """
    # Build store type to class mapping from DAQJobStoreConfig annotations
    try:
        resolved_hints = get_type_hints(DAQJobStoreConfig)
    except Exception:
        resolved_hints = {}

    store_type_to_class: dict[str, type] = {}
    for store_type, annotation in resolved_hints.items():
        # Get the actual type from the annotation (unwrap Optional)
        origin = get_origin(annotation)
        if origin is not None:
            args = get_args(annotation)
            non_none_args = [a for a in args if a is not type(None)]
            if non_none_args:
                config_class = non_none_args[0]
                store_type_to_class[store_type] = config_class

    templates: dict[str, dict] = {}

    for store_type, config_class in store_type_to_class.items():
        try:
            # Generate JSON schema using msgspec with custom hook for unsupported types
            schema = msgspec.json.schema(config_class, schema_hook=_schema_hook)
        except TypeError:
            # Skip types that can't be converted to schemas
            continue

        # Add our custom metadata
        schema["type_key"] = store_type
        schema["label"] = STORE_TYPE_LABELS.get(
            store_type, f"{store_type.upper()} Store"
        )
        schema["description"] = STORE_TYPE_DESCRIPTIONS.get(
            store_type, schema.get("description", f"Store data using {store_type}")
        )

        templates[store_type] = schema

    return templates

Generate store configuration templates using msgspec's JSON schema generation.

Returns a dictionary of JSON schemas for each store type.