API

API

Module for generating, sorting, and plotting data products.
This uses pydantic dataclasses for JSON serialization to avoid overloading system memory.

Some important learning material for pydantic classes and JSON (de)serialization:

• Nested Pydantic Models
• Deserializing Child Classes

Attributes:

Name	Type	Description
DATA_PRODUCTS_FNAME_DEFAULT	str	Hard-coded json file name 'data_products.json'

DATA_PRODUCTS_FNAME_DEFAULT module-attribute

DATA_PRODUCTS_FNAME_DEFAULT = 'data_products.json'

Hard-coded file name for storing data products in batch-processed input directories.

ProductGenerator module-attribute

ProductGenerator = Callable[[Path], ProductList]

Callable method type. Users must provide a ProductGenerator to map over raw data.

Parameters:

Name	Type	Description	Default
path	Path	Workdir holding raw data (Should be one per run from a batch)	required

Returns:

Type	Description
ProductList	List of data products to be sorted and used to produce assets

ProductList module-attribute

ProductList = List[SerializeAsAny[InstanceOf[DataProduct]]]

List of serializable DataProduct or child classes thereof

Tag module-attribute

Tag = Union[Tuple[Hashable, ...], Hashable]

Determines what types can be used to define a tag

Tags module-attribute

Tags = List[Tag]

List of tags

AxLine pydantic-model

Bases: PlottableData2D

Defines a horizontal or vertical line to be drawn on a plot.

Attributes:

Name	Type	Description
value	float	Value at which to draw the line (x-value for vertical, y-value for horizontal)
orientation	LineOrientation	Whether line should be horizontal or vertical
pen	Pen	Style and label information for drawing to matplotlib axes
tags	Tags	Tags to be used for sorting data
metadata	dict[str, str]	A dictionary of metadata

Show JSON schema:

{
  "$defs": {
    "Format2D": {
      "additionalProperties": false,
      "description": "Formatting data for matplotlib figure and axes\n\nAttributes:\n    title_fig (Optional[str]): Sets [figure title][matplotlib.figure.Figure.suptitle]\n    title_legend (Optional[str]): Sets [legend title][matplotlib.legend.Legend.set_title]\n    title_ax (Optional[str]): Sets [axis title][matplotlib.axes.Axes.set_title]\n    label_x (Optional[str]): Sets [x-axis label][matplotlib.axes.Axes.set_xlabel]\n    label_y (Optional[str]): Sets [y-axis label][matplotlib.axes.Axes.set_ylabel]\n    lim_x_min (float | str | None): Sets [x-axis lower bound][matplotlib.axes.Axes.set_xlim]\n    lim_x_max (float | str | None): Sets [x-axis upper bound][matplotlib.axes.Axes.set_xlim]\n    lim_y_min (float | str | None): Sets [y-axis lower bound][matplotlib.axes.Axes.set_ylim]\n    lim_y_max (float | str | None): Sets [y-axis upper bound][matplotlib.axes.Axes.set_ylim]",
      "properties": {
        "title_fig": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Title Fig"
        },
        "title_legend": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Title Legend"
        },
        "title_ax": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Title Ax"
        },
        "label_x": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Label X"
        },
        "label_y": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Label Y"
        },
        "lim_x_min": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Lim X Min"
        },
        "lim_x_max": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Lim X Max"
        },
        "lim_y_min": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Lim Y Min"
        },
        "lim_y_max": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Lim Y Max"
        }
      },
      "title": "Format2D",
      "type": "object"
    },
    "LineOrientation": {
      "description": "Defines orientation for axis lines\n\nAttributes:\n    HORIZONTAL (LineOrientation): Horizontal line\n    VERTICAL (LineOrientation): Vertical line",
      "enum": [
        "horizontal",
        "vertical"
      ],
      "title": "LineOrientation",
      "type": "string"
    },
    "Pen": {
      "additionalProperties": false,
      "description": "Defines the pen drawing to matplotlib.\n\nAttributes:\n    color (str): Color of line\n    size (float): Line width\n    alpha (float): Opacity from 0 to 1 (inclusive)\n    zorder (float): Prioritization \n    label (Union[str, None]): Legend label",
      "properties": {
        "color": {
          "default": "k",
          "title": "Color",
          "type": "string"
        },
        "size": {
          "default": 1,
          "title": "Size",
          "type": "number"
        },
        "alpha": {
          "default": 1,
          "title": "Alpha",
          "type": "number"
        },
        "zorder": {
          "default": 0,
          "title": "Zorder",
          "type": "number"
        },
        "label": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Label"
        }
      },
      "title": "Pen",
      "type": "object"
    }
  },
  "additionalProperties": false,
  "description": "Defines a horizontal or vertical line to be drawn on a plot.\n\nAttributes:\n    value (float): Value at which to draw the line (x-value for vertical, y-value for horizontal)\n    orientation (LineOrientation): Whether line should be horizontal or vertical\n    pen (Pen): Style and label information for drawing to matplotlib axes\n    tags (Tags): Tags to be used for sorting data\n    metadata (dict[str, str]): A dictionary of metadata",
  "properties": {
    "tags": {
      "items": {
        "anyOf": []
      },
      "title": "Tags",
      "type": "array"
    },
    "metadata": {
      "additionalProperties": {
        "type": "string"
      },
      "default": {},
      "title": "Metadata",
      "type": "object"
    },
    "format2d": {
      "anyOf": [
        {
          "$ref": "#/$defs/Format2D"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "value": {
      "title": "Value",
      "type": "number"
    },
    "orientation": {
      "$ref": "#/$defs/LineOrientation"
    },
    "pen": {
      "$ref": "#/$defs/Pen",
      "default": {
        "color": "k",
        "size": 1.0,
        "alpha": 1.0,
        "zorder": 0.0,
        "label": null
      }
    }
  },
  "required": [
    "tags",
    "value",
    "orientation"
  ],
  "title": "AxLine",
  "type": "object"
}

Config:

• extra: 'forbid'

Fields:

• value (float)
• orientation (LineOrientation)
• pen (Pen)

plot_to_ax

plot_to_ax(ax: Axes)

Plots line to matplotlib axes object.

Parameters:

Name	Type	Description	Default
ax	Axes	axes to which line should be plotted	required

Source code in src/trendify/API.py

def plot_to_ax(self, ax: plt.Axes):
    """
    Plots line to matplotlib axes object.

    Args:
        ax (plt.Axes): axes to which line should be plotted
    """
    match self.orientation:
        case LineOrientation.HORIZONTAL:
            ax.axhline(y=self.value, **self.pen.as_scatter_plot_kwargs())
        case LineOrientation.VERTICAL:
            ax.axvline(x=self.value, **self.pen.as_scatter_plot_kwargs())
        case _:
            print(f'Unrecognized line orientation {self.orientation}')

DataProduct pydantic-model

Bases: BaseModel

Base class for data products to be generated and handled.

Attributes:

Name	Type	Description
product_type	str	Product type should be the same as the class name. The product type is used to search for products from a DataProductCollection.
tags	Tags	Tags to be used for sorting data.
metadata	dict[str, str]	A dictionary of metadata to be used as a tool tip for mousover in grafana

Show JSON schema:

{
  "additionalProperties": true,
  "description": "Base class for data products to be generated and handled.\n\nAttributes:\n    product_type (str): Product type should be the same as the class name.\n        The product type is used to search for products from a [DataProductCollection][trendify.API.DataProductCollection].\n    tags (Tags): Tags to be used for sorting data.\n    metadata (dict[str, str]): A dictionary of metadata to be used as a tool tip for mousover in grafana",
  "properties": {
    "tags": {
      "items": {
        "anyOf": []
      },
      "title": "Tags",
      "type": "array"
    },
    "metadata": {
      "additionalProperties": {
        "type": "string"
      },
      "default": {},
      "title": "Metadata",
      "type": "object"
    }
  },
  "required": [
    "tags"
  ],
  "title": "DataProduct",
  "type": "object"
}

Config:

• extra: 'allow'

Fields:

• tags (Tags)
• metadata (dict[str, str])
• product_type (str)

Validators:

• _remove_computed_fields

product_type pydantic-field

product_type: str

Returns:

Type	Description
str	Product type should be the same as the class name. The product type is used to search for products from a DataProductCollection.

__init_subclass__

__init_subclass__(**kwargs: Any) -> None

Registers child subclasses to be able to parse them from JSON file using the deserialize_child_classes method

Source code in src/trendify/API.py

def __init_subclass__(cls, **kwargs: Any) -> None:
    """
    Registers child subclasses to be able to parse them from JSON file using the 
    [deserialize_child_classes][trendify.API.DataProduct.deserialize_child_classes] method
    """
    super().__init_subclass__(**kwargs)
    _data_product_subclass_registry[cls.__name__] = cls    

append_to_list

append_to_list(l: List)

Appends self to list.

Parameters:

Name	Type	Description	Default
l	List	list to which self will be appended	required

Returns:

Type	Description
Self	returns instance of self

Source code in src/trendify/API.py

def append_to_list(self, l: List):
    """
    Appends self to list.

    Args:
        l (List): list to which `self` will be appended

    Returns:
        (Self): returns instance of `self`
    """
    l.append(self)
    return self

deserialize_child_classes classmethod

deserialize_child_classes(key: str, **kwargs)

Loads json data to pydandic dataclass of whatever DataProduct child time is appropriate

Parameters:

Name	Type	Description	Default
key	str	json key	required
kwargs	dict	json entries stored under given key	{}

Source code in src/trendify/API.py

@classmethod
def deserialize_child_classes(cls, key: str, **kwargs):
    """
    Loads json data to pydandic dataclass of whatever DataProduct child time is appropriate

    Args:
        key (str): json key
        kwargs (dict): json entries stored under given key
    """
    type_key = 'product_type'
    elements = kwargs.get(key, None)
    if elements:
        for index in range(len(kwargs[key])):
            duck_info = kwargs[key][index]
            if isinstance(duck_info, dict):
                product_type = duck_info.pop(type_key)
                duck_type = _data_product_subclass_registry[product_type]
                kwargs[key][index] = duck_type(**duck_info)

DataProductCollection pydantic-model

DataProductCollection(**kwargs: Any)

Bases: BaseModel

A collection of data products.

Use this class to serialize data products to JSON, de-serialized them from JSON, filter the products, etc.

Attributes:

Name	Type	Description
elements	ProductList	A list of data products.

Show JSON schema:

{
  "$defs": {
    "DataProduct": {
      "additionalProperties": true,
      "description": "Base class for data products to be generated and handled.\n\nAttributes:\n    product_type (str): Product type should be the same as the class name.\n        The product type is used to search for products from a [DataProductCollection][trendify.API.DataProductCollection].\n    tags (Tags): Tags to be used for sorting data.\n    metadata (dict[str, str]): A dictionary of metadata to be used as a tool tip for mousover in grafana",
      "properties": {
        "tags": {
          "items": {
            "anyOf": []
          },
          "title": "Tags",
          "type": "array"
        },
        "metadata": {
          "additionalProperties": {
            "type": "string"
          },
          "default": {},
          "title": "Metadata",
          "type": "object"
        }
      },
      "required": [
        "tags"
      ],
      "title": "DataProduct",
      "type": "object"
    }
  },
  "description": "A collection of data products.\n\nUse this class to serialize data products to JSON, de-serialized them from JSON, filter the products, etc.\n\nAttributes:\n    elements (ProductList): A list of data products.",
  "properties": {
    "derived_from": {
      "anyOf": [
        {
          "format": "path",
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Derived From"
    },
    "elements": {
      "anyOf": [
        {
          "items": {
            "$ref": "#/$defs/DataProduct"
          },
          "type": "array"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Elements"
    }
  },
  "title": "DataProductCollection",
  "type": "object"
}

Fields:

• derived_from (Path | None)
• elements (ProductList | None)

Source code in src/trendify/API.py

def __init__(self, **kwargs: Any):
    DataProduct.deserialize_child_classes(key='elements', **kwargs)                
    super().__init__(**kwargs)

add_products

add_products(*products: DataProduct)

Parameters:

Name	Type	Description	Default
products	Tuple[DataProduct | ProductList, ...]	Products or lists of products to be appended to collection elements.	()

Source code in src/trendify/API.py

def add_products(self, *products: DataProduct):
    """
    Args:
        products (Tuple[DataProduct|ProductList, ...]): Products or lists of products to be
            appended to collection elements.  
    """
    self.elements.extend(flatten(products))

collect_from_all_jsons classmethod

collect_from_all_jsons(
    *dirs: Path, recursive: bool = False, data_products_filename: str | None = "*.json"
)

Loads all products from JSONs in the given list of directories.
If recursive is set to True, the directories will be searched recursively (this could lead to double counting if you pass in subdirectories of a parent).

Parameters:

Name	Type	Description	Default
dirs	Tuple[Path, ...]	Directories from which to load data product JSON files.	()
recursive	bool	whether or not to search each of the provided directories recursively for data product json files.	False

Returns:

Type	Description
Type[Self] | None	Data product collection if JSON files are found. Otherwise, returns None if no product JSON files were found.

Source code in src/trendify/API.py

@classmethod
def collect_from_all_jsons(cls, *dirs: Path, recursive: bool = False, data_products_filename: str|None = '*.json'):
    """
    Loads all products from JSONs in the given list of directories.  
    If recursive is set to `True`, the directories will be searched recursively 
    (this could lead to double counting if you pass in subdirectories of a parent).

    Args:
        dirs (Tuple[Path, ...]): Directories from which to load data product JSON files.
        recursive (bool): whether or not to search each of the provided directories recursively for 
            data product json files.

    Returns:
        (Type[Self] | None): Data product collection if JSON files are found.  
            Otherwise, returns None if no product JSON files were found.
    """
    if not recursive:
        jsons: List[Path] = list(flatten(chain(list(d.glob(data_products_filename)) for d in dirs)))
    else:
        jsons: List[Path] = list(flatten(chain(list(d.glob(f'**/{data_products_filename}')) for d in dirs)))
    if jsons:
        return cls.union(
            *tuple(
                [
                    cls.model_validate_json(p.read_text())
                    for p in jsons
                ]
            )
        )
    else:
        return None

drop_products

drop_products(tag: Tag | None = None, object_type: Type[R] | None = None) -> Self[R]

Removes products matching tag and/or object_type from collection elements.

Parameters:

Name	Type	Description	Default
tag	Tag | None	Tag for which data products should be dropped	None
object_type	Type | None	Type of data product to drop	None

Returns:

Type	Description
DataProductCollection	A new collection from which matching elements have been dropped.

Source code in src/trendify/API.py

def drop_products(self, tag: Tag | None = None, object_type: Type[R] | None = None) -> Self[R]:
    """
    Removes products matching `tag` and/or `object_type` from collection elements.

    Args:
        tag (Tag | None): Tag for which data products should be dropped
        object_type (Type | None): Type of data product to drop

    Returns:
        (DataProductCollection): A new collection from which matching elements have been dropped.
    """
    match_key = tag is None, object_type is None
    match match_key:
        case (True, True):
            return type(self)(elements=self.elements)
        case (True, False):
            return type(self)(elements=[e for e in self.elements if not isinstance(e, object_type)])
        case (False, True):
            return type(self)(elements=[e for e in self.elements if not tag in e.tags])
        case (False, False):
            return type(self)(elements=[e for e in self.elements if not (tag in e.tags and isinstance(e, object_type))])
        case _:
            raise ValueError('Something is wrong with match statement')

from_iterable classmethod

from_iterable(*products: Tuple[ProductList, ...])

Returns a new instance containing all of the products provided in the *products argument.

Parameters:

Name	Type	Description	Default
products	Tuple[ProductList, ...]	Lists of data products to combine into a collection	()

Returns:

Type	Description
cls	A data product collection containing all of the provided products in the *products argument.

Source code in src/trendify/API.py

@classmethod
def from_iterable(cls, *products: Tuple[ProductList, ...]):
    """
    Returns a new instance containing all of the products provided in the `*products` argument.

    Args:
        products (Tuple[ProductList, ...]): Lists of data products to combine into a collection

    Returns:
        (cls): A data product collection containing all of the provided products in the `*products` argument.
    """
    return cls(elements=list(flatten(products)))

get_products

get_products(tag: Tag | None = None, object_type: Type[R] | None = None) -> Self[R]

Returns a new collection containing products matching tag and/or object_type. Both tag and object_type default to None which matches all products.

Parameters:

Name	Type	Description	Default
tag	Tag | None	Tag of data products to be kept. None matches all products.	None
object_type	Type | None	Type of data product to keep. None matches all products.	None

Returns:

Type	Description
DataProductCollection	A new collection containing matching elements.

Source code in src/trendify/API.py

def get_products(self, tag: Tag | None = None, object_type: Type[R] | None = None) -> Self[R]:
    """
    Returns a new collection containing products matching `tag` and/or `object_type`.
    Both `tag` and `object_type` default to `None` which matches all products.

    Args:
        tag (Tag | None): Tag of data products to be kept.  `None` matches all products.
        object_type (Type | None): Type of data product to keep.  `None` matches all products.

    Returns:
        (DataProductCollection): A new collection containing matching elements.
    """
    match_key = tag is None, object_type is None
    match match_key:
        case (True, True):
            return type(self)(elements=self.elements)
        case (True, False):
            return type(self)(elements=[e for e in self.elements if isinstance(e, object_type)])
        case (False, True):
            return type(self)(elements=[e for e in self.elements if tag in e.tags])
        case (False, False):
            return type(self)(elements=[e for e in self.elements if tag in e.tags and isinstance(e, object_type)])
        case _:
            raise ValueError('Something is wrong with match statement')

get_tags

get_tags(data_product_type: Type[DataProduct] | None = None) -> set

Gets the tags related to a given type of DataProduct. Parent classes will match all child class types.

Parameters:

Name	Type	Description	Default
data_product_type	Type[DataProduct] | None	type for which you want to get the list of tags	None

Returns:

Type	Description
set	set of tags applying to the given data_product_type.

Source code in src/trendify/API.py

def get_tags(self, data_product_type: Type[DataProduct] | None = None) -> set:
    """
    Gets the tags related to a given type of `DataProduct`.  Parent classes will match all child class types.

    Args:
        data_product_type (Type[DataProduct] | None): type for which you want to get the list of tags

    Returns:
        (set): set of tags applying to the given `data_product_type`.
    """
    tags = []
    for e in flatten(self.elements):
        if data_product_type is None or isinstance(e, data_product_type):
            for t in e.tags:
                tags.append(t)
    return set(tags)

make_grafana_panels classmethod

make_grafana_panels(dir_in: Path, panel_dir: Path, server_path: str)

Processes collection of elements corresponding to a single tag. This method should be called on a directory containing jsons for which the products have been sorted.

Parameters:

Name	Type	Description	Default
dir_in	Path	Directory from which to read data products (should be sorted first)	required
panel_dir	Path	Where to put the panel information	required

Source code in src/trendify/API.py

@classmethod
def make_grafana_panels(
        cls,
        dir_in: Path,
        panel_dir: Path,
        server_path: str,
    ):
    """
    Processes collection of elements corresponding to a single tag.
    This method should be called on a directory containing jsons for which the products have been
    sorted.

    Args:
        dir_in (Path): Directory from which to read data products (should be sorted first)
        panel_dir (Path): Where to put the panel information
    """
    import grafana_api as gapi
    collection = cls.collect_from_all_jsons(dir_in)
    panel_dir.mkdir(parents=True, exist_ok=True)

    if collection is not None:
        for tag in collection.get_tags():
            dot_tag = '.'.join([str(t) for t in tag]) if should_be_flattened(tag) else tag
            underscore_tag = '_'.join([str(t) for t in tag]) if should_be_flattened(tag) else tag

            table_entries: List[TableEntry] = collection.get_products(tag=tag, object_type=TableEntry).elements

            if table_entries:
                print(f'\n\nMaking tables for {tag = }\n')
                panel = gapi.Panel(
                    title=str(tag).capitalize() if isinstance(tag, str) else ' '.join([str(t).title() for t in tag]),
                    targets=[
                        gapi.Target(
                            datasource=gapi.DataSource(),
                            url='/'.join([server_path.strip('/'), dot_tag, 'TableEntry']),
                            uql=UQL_TableEntry,
                        )
                    ],
                    type='table',
                )
                panel_dir.joinpath(underscore_tag + '_table_panel.json').write_text(panel.model_dump_json())
                print(f'\nFinished tables for {tag = }\n')

            traces: List[Trace2D] = collection.get_products(tag=tag, object_type=Trace2D).elements
            points: List[Point2D] = collection.get_products(tag=tag, object_type=Point2D).elements

            if points or traces:
                print(f'\n\nMaking xy chart for {tag = }\n')
                panel = gapi.Panel(
                    targets=[
                        gapi.Target(
                            datasource=gapi.DataSource(),
                            url='/'.join([server_path.strip('/'), dot_tag, 'Point2D']),
                            uql=UQL_Point2D,
                            refId='A',
                        ),
                        gapi.Target(
                            datasource=gapi.DataSource(),
                            url='/'.join([server_path.strip('/'), dot_tag, 'Trace2D']),
                            uql=UQL_Trace2D,
                            refId='B',
                        )
                    ],
                    transformations=[
                        gapi.Merge(),
                        gapi.PartitionByValues.from_fields(
                            fields='label',
                            keep_fields=False,
                            fields_as_labels=False,
                        )
                    ],
                    type='xychart',
                )
                panel_dir.joinpath(underscore_tag + '_xy_panel.json').write_text(panel.model_dump_json())
                print(f'\nFinished xy plot for {tag = }\n')

process_collection classmethod

process_collection(
    dir_in: Path, dir_out: Path, no_tables: bool, no_xy_plots: bool, no_histograms: bool, dpi: int
)

Processes collection of elements corresponding to a single tag. This method should be called on a directory containing jsons for which the products have been sorted.

Parameters:

Name	Type	Description	Default
dir_in	Path	Input directory for loading assets	required
dir_out	Path	Output directory for assets	required
no_tables	bool	Suppresses table asset creation	required
no_xy_plots	bool	Suppresses xy plot asset creation	required
no_histograms	bool	Suppresses histogram asset creation	required
dpi	int	Sets resolution of asset output	required

Source code in src/trendify/API.py

@classmethod
def process_collection(
        cls,
        dir_in: Path,
        dir_out: Path,
        no_tables: bool,
        no_xy_plots: bool,
        no_histograms: bool,
        dpi: int,
    ):
    """
    Processes collection of elements corresponding to a single tag.
    This method should be called on a directory containing jsons for which the products have been
    sorted.

    Args:
        dir_in (Path):  Input directory for loading assets
        dir_out (Path):  Output directory for assets
        no_tables (bool):  Suppresses table asset creation
        no_xy_plots (bool):  Suppresses xy plot asset creation
        no_histograms (bool):  Suppresses histogram asset creation
        dpi (int):  Sets resolution of asset output
    """

    collection = cls.collect_from_all_jsons(dir_in)

    if collection is not None:

        for tag in collection.get_tags():
        # tags = collection.get_tags()
        # try:
        #     [tag] = collection.get_tags()
        # except:
        #     breakpoint()

            if not no_tables:

                table_entries: List[TableEntry] = collection.get_products(tag=tag, object_type=TableEntry).elements

                if table_entries:
                    print(f'\n\nMaking tables for {tag = }\n')
                    TableBuilder.process_table_entries(
                        tag=tag,
                        table_entries=table_entries,
                        out_dir=dir_out
                    )
                    print(f'\nFinished tables for {tag = }\n')

            if not no_xy_plots:

                traces: List[Trace2D] = collection.get_products(tag=tag, object_type=Trace2D).elements
                points: List[Point2D] = collection.get_products(tag=tag, object_type=Point2D).elements
                axlines: List[AxLine] = collection.get_products(tag=tag, object_type=AxLine).elements  # Add this line

                if points or traces or axlines:  # Update condition
                    print(f'\n\nMaking xy plot for {tag = }\n')
                    XYDataPlotter.handle_points_and_traces(
                        tag=tag,
                        points=points,
                        traces=traces,
                        axlines=axlines,  # Add this parameter
                        dir_out=dir_out,
                        dpi=dpi,
                    )
                    print(f'\nFinished xy plot for {tag = }\n')

                # traces: List[Trace2D] = collection.get_products(tag=tag, object_type=Trace2D).elements
                # points: List[Point2D] = collection.get_products(tag=tag, object_type=Point2D).elements
                # if points or traces:
                #     print(f'\n\nMaking xy plot for {tag = }\n')
                #     XYDataPlotter.handle_points_and_traces(
                #         tag=tag,
                #         points=points,
                #         traces=traces,
                #         dir_out=dir_out,
                #         dpi=dpi,
                #     )
                #     print(f'\nFinished xy plot for {tag = }\n')

            if not no_histograms:
                histogram_entries: List[HistogramEntry] = collection.get_products(tag=tag, object_type=HistogramEntry).elements

                if histogram_entries:
                    print(f'\n\nMaking histogram for {tag = }\n')
                    Histogrammer.handle_histogram_entries(
                        tag=tag,
                        histogram_entries=histogram_entries,
                        dir_out=dir_out,
                        dpi=dpi
                    )
                    print(f'\nFinished histogram for {tag = }\n')

sort_by_tags classmethod

sort_by_tags(
    dirs_in: List[Path], dir_out: Path, data_products_fname: str = DATA_PRODUCTS_FNAME_DEFAULT
)

Loads the data product JSON files from dirs_in sorts the products. Sorted products are written to smaller files in a nested directory structure under dir_out. A nested directory structure is generated according to the data tags. Resulting product files are named according to the directory from which they were originally loaded.

Parameters:

Name	Type	Description	Default
dirs_in	List[Path]	Directories from which the data product JSON files are to be loaded.	required
dir_out	Path	Directory to which the sorted data products will be written into a nested folder structure generated according to the data tags.	required
data_products_fname	str	Name of data products file	DATA_PRODUCTS_FNAME_DEFAULT

Source code in src/trendify/API.py

@classmethod
def sort_by_tags(cls, dirs_in: List[Path], dir_out: Path, data_products_fname: str = DATA_PRODUCTS_FNAME_DEFAULT):
    """
    Loads the data product JSON files from `dirs_in` sorts the products.
    Sorted products are written to smaller files in a nested directory structure under `dir_out`.
    A nested directory structure is generated according to the data tags.
    Resulting product files are named according to the directory from which they were originally loaded.

    Args:
        dirs_in (List[Path]): Directories from which the data product JSON files are to be loaded.
        dir_out (Path): Directory to which the sorted data products will be written into a 
            nested folder structure generated according to the data tags.
        data_products_fname (str): Name of data products file
    """
    dirs_in = list(dirs_in)
    dirs_in.sort()
    len_dirs = len(dirs_in)
    for n, dir_in in enumerate(dirs_in):
        print(f'Sorting tagged data from dir {n}/{len_dirs}', end=f'\r')
        cls.sort_by_tags_single_directory(dir_in=dir_in, dir_out=dir_out, data_products_fname=data_products_fname)

sort_by_tags_single_directory classmethod

sort_by_tags_single_directory(
    dir_in: Path, dir_out: Path, data_products_fname: str = DATA_PRODUCTS_FNAME_DEFAULT
)

Loads the data product JSON files from dir_in and sorts the products. Sorted products are written to smaller files in a nested directory structure under dir_out. A nested directory structure is generated according to the data tags. Resulting product files are named according to the directory from which they were originally loaded.

Parameters:

Name	Type	Description	Default
dir_in	List[Path]	Directories from which the data product JSON files are to be loaded.	required
dir_out	Path	Directory to which the sorted data products will be written into a nested folder structure generated according to the data tags.	required
data_products_fname	str	Name of data products file	DATA_PRODUCTS_FNAME_DEFAULT

Source code in src/trendify/API.py

@classmethod
def sort_by_tags_single_directory(cls, dir_in: Path, dir_out: Path, data_products_fname: str = DATA_PRODUCTS_FNAME_DEFAULT):
    """
    Loads the data product JSON files from `dir_in` and sorts the products.
    Sorted products are written to smaller files in a nested directory structure under `dir_out`.
    A nested directory structure is generated according to the data tags.
    Resulting product files are named according to the directory from which they were originally loaded.

    Args:
        dir_in (List[Path]): Directories from which the data product JSON files are to be loaded.
        dir_out (Path): Directory to which the sorted data products will be written into a 
            nested folder structure generated according to the data tags.
        data_products_fname (str): Name of data products file
    """
    products_file = dir_in.joinpath(data_products_fname)
    if products_file.exists():
        print(f'Sorting results from {dir_in = }')
        collection = DataProductCollection.model_validate_json(dir_in.joinpath(data_products_fname).read_text())
        collection.derived_from = dir_in
        tags = collection.get_tags()
        for tag in tags:
            sub_collection = collection.get_products(tag=tag)
            save_dir = dir_out.joinpath(*atleast_1d(tag))
            save_dir.mkdir(parents=True, exist_ok=True)
            next_index = get_and_reserve_next_index(save_dir=save_dir, dir_in=dir_in)
            file = save_dir.joinpath(str(next_index)).with_suffix('.json')
            file.write_text(sub_collection.model_dump_json())
    else:
        print(f'No results found in {dir_in = }')

union classmethod

union(*collections: DataProductCollection)

Aggregates all of the products from multiple collections into a new larger collection.

Parameters:

Name	Type	Description	Default
collections	Tuple[DataProductCollection, ...]	Data product collections for which the products should be combined into a new collection.	()

Returns:

Type	Description
Type[Self]	A new data product collection containing all products from the provided *collections.

Source code in src/trendify/API.py

@classmethod
def union(cls, *collections: DataProductCollection):
    """
    Aggregates all of the products from multiple collections into a new larger collection.

    Args:
        collections (Tuple[DataProductCollection, ...]): Data product collections
            for which the products should be combined into a new collection.

    Returns:
        (Type[Self]): A new data product collection containing all products from
            the provided `*collections`.
    """
    return cls(elements=list(flatten(chain(c.elements for c in collections))))

DataProductGenerator

DataProductGenerator(processor: ProductGenerator)

A wrapper for saving the data products generated by a user defined function

Parameters:

Name	Type	Description	Default
processor	ProductGenerator	A callable that receives a working directory and returns a list of data products.	required

Source code in src/trendify/API.py

def __init__(self, processor: ProductGenerator):
    self._processor = processor

process_and_save

process_and_save(workdir: Path, data_products_fname: str = DATA_PRODUCTS_FNAME_DEFAULT)

Runs the user-defined processor method stored at instantiation.

Saves the returned products to a JSON file in the same directory.

Parameters:

Name	Type	Description	Default
workdir	Path	working directory on which to run the processor method.	required
data_products_fname	str	Name of data products file	DATA_PRODUCTS_FNAME_DEFAULT

Source code in src/trendify/API.py

def process_and_save(self, workdir: Path, data_products_fname: str = DATA_PRODUCTS_FNAME_DEFAULT):
    """
    Runs the user-defined processor method stored at instantiation.

    Saves the returned products to a JSON file in the same directory.

    Args:
        workdir (Path): working directory on which to run the processor method.
        data_products_fname (str): Name of data products file
    """

    print(f'Processing {workdir = } with {self._processor = }')
    collection = DataProductCollection.from_iterable(self._processor(workdir))
    if collection.elements:
        workdir.mkdir(exist_ok=True, parents=True)
        workdir.joinpath(data_products_fname).write_text(collection.model_dump_json())

Format2D pydantic-model

Bases: HashableBase

Formatting data for matplotlib figure and axes

Attributes:

Name	Type	Description
title_fig	Optional[str]	Sets figure title
title_legend	Optional[str]	Sets legend title
title_ax	Optional[str]	Sets axis title
label_x	Optional[str]	Sets x-axis label
label_y	Optional[str]	Sets y-axis label
lim_x_min	float | str | None	Sets x-axis lower bound
lim_x_max	float | str | None	Sets x-axis upper bound
lim_y_min	float | str | None	Sets y-axis lower bound
lim_y_max	float | str | None	Sets y-axis upper bound

Show JSON schema:

{
  "additionalProperties": false,
  "description": "Formatting data for matplotlib figure and axes\n\nAttributes:\n    title_fig (Optional[str]): Sets [figure title][matplotlib.figure.Figure.suptitle]\n    title_legend (Optional[str]): Sets [legend title][matplotlib.legend.Legend.set_title]\n    title_ax (Optional[str]): Sets [axis title][matplotlib.axes.Axes.set_title]\n    label_x (Optional[str]): Sets [x-axis label][matplotlib.axes.Axes.set_xlabel]\n    label_y (Optional[str]): Sets [y-axis label][matplotlib.axes.Axes.set_ylabel]\n    lim_x_min (float | str | None): Sets [x-axis lower bound][matplotlib.axes.Axes.set_xlim]\n    lim_x_max (float | str | None): Sets [x-axis upper bound][matplotlib.axes.Axes.set_xlim]\n    lim_y_min (float | str | None): Sets [y-axis lower bound][matplotlib.axes.Axes.set_ylim]\n    lim_y_max (float | str | None): Sets [y-axis upper bound][matplotlib.axes.Axes.set_ylim]",
  "properties": {
    "title_fig": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Title Fig"
    },
    "title_legend": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Title Legend"
    },
    "title_ax": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Title Ax"
    },
    "label_x": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Label X"
    },
    "label_y": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Label Y"
    },
    "lim_x_min": {
      "anyOf": [
        {
          "type": "number"
        },
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Lim X Min"
    },
    "lim_x_max": {
      "anyOf": [
        {
          "type": "number"
        },
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Lim X Max"
    },
    "lim_y_min": {
      "anyOf": [
        {
          "type": "number"
        },
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Lim Y Min"
    },
    "lim_y_max": {
      "anyOf": [
        {
          "type": "number"
        },
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Lim Y Max"
    }
  },
  "title": "Format2D",
  "type": "object"
}

Config:

• extra: 'forbid'

Fields:

• title_fig (Optional[str] | None)
• title_legend (Optional[str] | None)
• title_ax (Optional[str] | None)
• label_x (Optional[str] | None)
• label_y (Optional[str] | None)
• lim_x_min (float | str | None)
• lim_x_max (float | str | None)
• lim_y_min (float | str | None)
• lim_y_max (float | str | None)

union_from_iterable classmethod

union_from_iterable(format2ds: Iterable[Format2D])

Gets the most inclusive format object (in terms of limits) from a list of Format2D objects. Requires that the label and title fields are identical for all format objects in the list.

Parameters:

Name	Type	Description	Default
format2ds	Iterable[Format2D]	Iterable of Format2D objects.	required

Returns:

Type	Description
Format2D	Single format object from list of objects.

Source code in src/trendify/API.py

@classmethod
def union_from_iterable(cls, format2ds: Iterable[Format2D]):
    """
    Gets the most inclusive format object (in terms of limits) from a list of `Format2D` objects.
    Requires that the label and title fields are identical for all format objects in the list.

    Args:
        format2ds (Iterable[Format2D]): Iterable of `Format2D` objects.

    Returns:
        (Format2D): Single format object from list of objects.

    """
    formats = list(set(format2ds) - {None})
    [title_fig] = set(i.title_fig for i in formats if i is not None)
    [title_legend] = set(i.title_legend for i in formats if i is not None)
    [title_ax] = set(i.title_ax for i in formats if i is not None)
    [label_x] = set(i.label_x for i in formats if i is not None)
    [label_y] = set(i.label_y for i in formats if i is not None)
    x_min = [i.lim_x_min for i in formats if i.lim_x_min is not None]
    x_max = [i.lim_x_max for i in formats if i.lim_x_max is not None]
    y_min = [i.lim_y_min for i in formats if i.lim_y_min is not None]
    y_max = [i.lim_y_max for i in formats if i.lim_y_max is not None]
    lim_x_min = np.min(x_min) if len(x_min) > 0 else None
    lim_x_max = np.max(x_max) if len(x_max) > 0 else None
    lim_y_min = np.min(y_min) if len(y_min) > 0 else None
    lim_y_max = np.max(y_max) if len(y_max) > 0 else None

    return cls(
        title_fig=title_fig,
        title_legend=title_legend,
        title_ax=title_ax,
        label_x=label_x,
        label_y=label_y,
        lim_x_min=lim_x_min,
        lim_x_max=lim_x_max,
        lim_y_min=lim_y_min,
        lim_y_max=lim_y_max,
    )

HashableBase pydantic-model

Bases: BaseModel

Defines a base for hashable pydantic data classes so that they can be reduced to a minimal set through type-casting.

Show JSON schema:

{
  "description": "Defines a base for hashable pydantic data classes so that they can be reduced to a minimal set through type-casting.",
  "properties": {},
  "title": "HashableBase",
  "type": "object"
}

__hash__

__hash__()

Defines hash function

Source code in src/trendify/API.py

def __hash__(self):
    """
    Defines hash function
    """
    return hash((type(self),) + tuple(self.__dict__.values()))

HistogramEntry pydantic-model

Bases: PlottableData2D

Use this class to specify a value to be collected into a matplotlib histogram.

Attributes:

Name	Type	Description
tags	Tags	Tags used to sort data products
value	float | str	Value to be binned
style	HistogramStyle	Style of histogram display

Show JSON schema:

{
  "$defs": {
    "Format2D": {
      "additionalProperties": false,
      "description": "Formatting data for matplotlib figure and axes\n\nAttributes:\n    title_fig (Optional[str]): Sets [figure title][matplotlib.figure.Figure.suptitle]\n    title_legend (Optional[str]): Sets [legend title][matplotlib.legend.Legend.set_title]\n    title_ax (Optional[str]): Sets [axis title][matplotlib.axes.Axes.set_title]\n    label_x (Optional[str]): Sets [x-axis label][matplotlib.axes.Axes.set_xlabel]\n    label_y (Optional[str]): Sets [y-axis label][matplotlib.axes.Axes.set_ylabel]\n    lim_x_min (float | str | None): Sets [x-axis lower bound][matplotlib.axes.Axes.set_xlim]\n    lim_x_max (float | str | None): Sets [x-axis upper bound][matplotlib.axes.Axes.set_xlim]\n    lim_y_min (float | str | None): Sets [y-axis lower bound][matplotlib.axes.Axes.set_ylim]\n    lim_y_max (float | str | None): Sets [y-axis upper bound][matplotlib.axes.Axes.set_ylim]",
      "properties": {
        "title_fig": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Title Fig"
        },
        "title_legend": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Title Legend"
        },
        "title_ax": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Title Ax"
        },
        "label_x": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Label X"
        },
        "label_y": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Label Y"
        },
        "lim_x_min": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Lim X Min"
        },
        "lim_x_max": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Lim X Max"
        },
        "lim_y_min": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Lim Y Min"
        },
        "lim_y_max": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Lim Y Max"
        }
      },
      "title": "Format2D",
      "type": "object"
    },
    "HistogramStyle": {
      "description": "Label and style data for generating histogram bars\n\nAttributes:\n    color (str): Color of bars\n    label (str|None): Legend entry\n    histtype (str): Histogram type corresponding to matplotlib argument of same name\n    alpha_edge (float): Opacity of bar edge\n    alpha_face (float): Opacity of bar face\n    linewidth (float): Line width of bar outline\n    bins (int | list[int] | Tuple[int] | NDArray[Shape[\"*\"], int] | None): Number of bins (see [matplotlib docs][matplotlib.pyplot.hist])",
      "properties": {
        "color": {
          "default": "k",
          "title": "Color",
          "type": "string"
        },
        "label": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Label"
        },
        "histtype": {
          "default": "stepfilled",
          "title": "Histtype",
          "type": "string"
        },
        "alpha_edge": {
          "default": 1,
          "title": "Alpha Edge",
          "type": "number"
        },
        "alpha_face": {
          "default": 0.3,
          "title": "Alpha Face",
          "type": "number"
        },
        "linewidth": {
          "default": 2,
          "title": "Linewidth",
          "type": "number"
        },
        "bins": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "items": {
                "type": "integer"
              },
              "type": "array"
            },
            {
              "maxItems": 1,
              "minItems": 1,
              "prefixItems": [
                {
                  "type": "integer"
                }
              ],
              "type": "array"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Bins"
        }
      },
      "title": "HistogramStyle",
      "type": "object"
    }
  },
  "additionalProperties": false,
  "description": "Use this class to specify a value to be collected into a matplotlib histogram.\n\nAttributes:\n    tags (Tags): Tags used to sort data products\n    value (float | str): Value to be binned\n    style (HistogramStyle): Style of histogram display",
  "properties": {
    "tags": {
      "items": {
        "anyOf": []
      },
      "title": "Tags",
      "type": "array"
    },
    "metadata": {
      "additionalProperties": {
        "type": "string"
      },
      "default": {},
      "title": "Metadata",
      "type": "object"
    },
    "format2d": {
      "anyOf": [
        {
          "$ref": "#/$defs/Format2D"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "value": {
      "anyOf": [
        {
          "type": "number"
        },
        {
          "type": "string"
        }
      ],
      "title": "Value"
    },
    "style": {
      "anyOf": [
        {
          "$ref": "#/$defs/HistogramStyle"
        },
        {
          "type": "null"
        }
      ]
    }
  },
  "required": [
    "tags",
    "value"
  ],
  "title": "HistogramEntry",
  "type": "object"
}

Config:

• extra: 'forbid'

Fields:

• value (float | str)
• tags (Tags)
• style (HistogramStyle | None)

HistogramStyle pydantic-model

Bases: HashableBase

Label and style data for generating histogram bars

Attributes:

Name	Type	Description
color	str	Color of bars
label	str | None	Legend entry
histtype	str	Histogram type corresponding to matplotlib argument of same name
alpha_edge	float	Opacity of bar edge
alpha_face	float	Opacity of bar face
linewidth	float	Line width of bar outline
bins	int | list[int] | Tuple[int] | NDArray[Shape['*'], int] | None	Number of bins (see matplotlib docs)

Show JSON schema:

{
  "description": "Label and style data for generating histogram bars\n\nAttributes:\n    color (str): Color of bars\n    label (str|None): Legend entry\n    histtype (str): Histogram type corresponding to matplotlib argument of same name\n    alpha_edge (float): Opacity of bar edge\n    alpha_face (float): Opacity of bar face\n    linewidth (float): Line width of bar outline\n    bins (int | list[int] | Tuple[int] | NDArray[Shape[\"*\"], int] | None): Number of bins (see [matplotlib docs][matplotlib.pyplot.hist])",
  "properties": {
    "color": {
      "default": "k",
      "title": "Color",
      "type": "string"
    },
    "label": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Label"
    },
    "histtype": {
      "default": "stepfilled",
      "title": "Histtype",
      "type": "string"
    },
    "alpha_edge": {
      "default": 1,
      "title": "Alpha Edge",
      "type": "number"
    },
    "alpha_face": {
      "default": 0.3,
      "title": "Alpha Face",
      "type": "number"
    },
    "linewidth": {
      "default": 2,
      "title": "Linewidth",
      "type": "number"
    },
    "bins": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "items": {
            "type": "integer"
          },
          "type": "array"
        },
        {
          "maxItems": 1,
          "minItems": 1,
          "prefixItems": [
            {
              "type": "integer"
            }
          ],
          "type": "array"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Bins"
    }
  },
  "title": "HistogramStyle",
  "type": "object"
}

Fields:

• color (str)
• label (str | None)
• histtype (str)
• alpha_edge (float)
• alpha_face (float)
• linewidth (float)
• bins (int | list[int] | Tuple[int] | NDArray[Shape['*'], int] | None)

as_plot_kwargs

as_plot_kwargs()

Returns:

Type	Description
dict	kwargs for matplotlib hist method

Source code in src/trendify/API.py

def as_plot_kwargs(self):
    """
    Returns:
        (dict): kwargs for matplotlib `hist` method
    """
    return {
        'facecolor': (self.color, self.alpha_face),
        'edgecolor': (self.color, self.alpha_edge),
        'linewidth': self.linewidth,
        'label': self.label,
        'histtype': self.histtype,
        'bins': self.bins,
    }

Histogrammer

Histogrammer(in_dirs: List[Path], out_dir: Path, dpi: int)

Class for loading data products and histogramming the HistogramEntrys

Parameters:

Name	Type	Description	Default
in_dirs	List[Path]	Directories from which the data products are to be loaded.	required
out_dir	Path	Directory to which the generated histogram will be stored	required
dpi	int	resolution of plot	required

Source code in src/trendify/API.py

def __init__(
        self,
        in_dirs: List[Path],
        out_dir: Path,
        dpi: int,
    ):
    self.in_dirs = in_dirs
    self.out_dir = out_dir
    self.dpi = dpi

handle_histogram_entries classmethod

handle_histogram_entries(
    tag: Tag, histogram_entries: List[HistogramEntry], dir_out: Path, dpi: int
)

Histograms the provided entries. Formats and saves the figure. Closes the figure.

Parameters:

Name	Type	Description	Default
tag	Tag	Tag used to filter the loaded data products	required
histogram_entries	List[HistogramEntry]	A list of HistogramEntrys	required
dir_out	Path	Directory to which the generated histogram will be stored	required
dpi	int	resolution of plot	required

Source code in src/trendify/API.py

@classmethod
def handle_histogram_entries(
        cls, 
        tag: Tag, 
        histogram_entries: List[HistogramEntry],
        dir_out: Path,
        dpi: int,
    ):
    """
    Histograms the provided entries. Formats and saves the figure.  Closes the figure.

    Args:
        tag (Tag): Tag used to filter the loaded data products
        histogram_entries (List[HistogramEntry]): A list of [`HistogramEntry`][trendify.API.HistogramEntry]s
        dir_out (Path): Directory to which the generated histogram will be stored
        dpi (int): resolution of plot
    """
    saf = SingleAxisFigure.new(tag=tag)

    histogram_styles = set([h.style for h in histogram_entries])
    for s in histogram_styles:
        matching_entries = [e for e in histogram_entries if e.style == s]
        values = [e.value for e in matching_entries]
        if s is not None:
            saf.ax.hist(values, **s.as_plot_kwargs())
        else:
            saf.ax.hist(values)

    try:
        format2d_set = set([h.format2d for h in histogram_entries]) - {None}
        [format2d] = format2d_set
        saf.apply_format(format2d=format2d)
    except:
        print(f'Format not applied to {save_path  = } multiple entries conflict for given tag:\n\t{format2d_set = }')
    save_path = dir_out.joinpath(*tuple(atleast_1d(tag))).with_suffix('.jpg')
    save_path.parent.mkdir(exist_ok=True, parents=True)
    print(f'Saving to {save_path}')
    saf.savefig(save_path, dpi=dpi)
    del saf

plot

plot(tag: Tag, data_products_fname: str = DATA_PRODUCTS_FNAME_DEFAULT)

Generates a histogram by loading data from stored in_dirs and saves the plot to out_dir directory. A nested folder structure will be created if the provided tag is a tuple.
In that case, the last tag item (with an appropriate suffix) will be used for the file name.

Parameters:

Name	Type	Description	Default
tag	Tag	Tag used to filter the loaded data products	required

Source code in src/trendify/API.py

def plot(
        self,
        tag: Tag,
        data_products_fname: str = DATA_PRODUCTS_FNAME_DEFAULT,
    ):
    """
    Generates a histogram by loading data from stored `in_dirs` and saves the plot to `out_dir` directory.
    A nested folder structure will be created if the provided `tag` is a tuple.  
    In that case, the last tag item (with an appropriate suffix) will be used for the file name.

    Args:
        tag (Tag): Tag used to filter the loaded data products
    """
    print(f'Making histogram plot for {tag = }')

    histogram_entries: List[HistogramEntry] = []
    for directory in self.in_dirs:
        collection = DataProductCollection.model_validate_json(directory.joinpath(data_products_fname).read_text())
        histogram_entries.extend(collection.get_products(tag=tag, object_type=HistogramEntry).elements)

    self.handle_histogram_entries(
        tag=tag,
        histogram_entries=histogram_entries,
        dir_out=self.out_dir,
        dpi=self.dpi,
    )

LineOrientation

Bases: Enum

Defines orientation for axis lines

Attributes:

Name	Type	Description
HORIZONTAL	LineOrientation	Horizontal line
VERTICAL	LineOrientation	Vertical line

Marker pydantic-model

Bases: HashableBase

Defines marker for scattering to matplotlib

Attributes:

Name	Type	Description
color	str	Color of line
size	float	Line width
alpha	float	Opacity from 0 to 1 (inclusive)
zorder	float	Prioritization
label	Union[str, None]	Legend label
symbol	str	Matplotlib symbol string

Show JSON schema:

{
  "additionalProperties": false,
  "description": "Defines marker for scattering to matplotlib\n\nAttributes:\n    color (str): Color of line\n    size (float): Line width\n    alpha (float): Opacity from 0 to 1 (inclusive)\n    zorder (float): Prioritization \n    label (Union[str, None]): Legend label\n    symbol (str): Matplotlib symbol string",
  "properties": {
    "color": {
      "default": "k",
      "title": "Color",
      "type": "string"
    },
    "size": {
      "default": 5,
      "title": "Size",
      "type": "number"
    },
    "alpha": {
      "default": 1,
      "title": "Alpha",
      "type": "number"
    },
    "zorder": {
      "default": 0,
      "title": "Zorder",
      "type": "number"
    },
    "label": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Label"
    },
    "symbol": {
      "default": ".",
      "title": "Symbol",
      "type": "string"
    }
  },
  "title": "Marker",
  "type": "object"
}

Config:

• extra: 'forbid'

Fields:

• color (str)
• size (float)
• alpha (float)
• zorder (float)
• label (str | None)
• symbol (str)

as_scatter_plot_kwargs

as_scatter_plot_kwargs()

Returns:

Type	Description
dict	dictionary of kwargs for matplotlib scatter

Source code in src/trendify/API.py

def as_scatter_plot_kwargs(self):
    """
    Returns:
        (dict): dictionary of `kwargs` for [matplotlib scatter][matplotlib.axes.Axes.scatter]
    """
    return {
        'marker': self.symbol,
        'c': self.color,
        's': self.size,
        'alpha': self.alpha,
        'zorder': self.zorder,
        'label': self.label,
        'marker': self.symbol,
    }

from_pen classmethod

from_pen(pen: Pen, symbol: str = '.')

Converts Pen to marker with the option to specify a symbol

Source code in src/trendify/API.py

@classmethod
def from_pen(
        cls,
        pen: Pen,
        symbol: str = '.',
    ):
    """
    Converts Pen to marker with the option to specify a symbol
    """
    return cls(symbol=symbol, **pen.model_dump())

Pen pydantic-model

Bases: HashableBase

Defines the pen drawing to matplotlib.

Attributes:

Name	Type	Description
color	str	Color of line
size	float	Line width
alpha	float	Opacity from 0 to 1 (inclusive)
zorder	float	Prioritization
label	Union[str, None]	Legend label

Show JSON schema:

{
  "additionalProperties": false,
  "description": "Defines the pen drawing to matplotlib.\n\nAttributes:\n    color (str): Color of line\n    size (float): Line width\n    alpha (float): Opacity from 0 to 1 (inclusive)\n    zorder (float): Prioritization \n    label (Union[str, None]): Legend label",
  "properties": {
    "color": {
      "default": "k",
      "title": "Color",
      "type": "string"
    },
    "size": {
      "default": 1,
      "title": "Size",
      "type": "number"
    },
    "alpha": {
      "default": 1,
      "title": "Alpha",
      "type": "number"
    },
    "zorder": {
      "default": 0,
      "title": "Zorder",
      "type": "number"
    },
    "label": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Label"
    }
  },
  "title": "Pen",
  "type": "object"
}

Config:

• extra: 'forbid'

Fields:

• color (str)
• size (float)
• alpha (float)
• zorder (float)
• label (Union[str, None])

as_scatter_plot_kwargs

as_scatter_plot_kwargs()

Returns kwargs dictionary for passing to matplotlib plot method

Source code in src/trendify/API.py

def as_scatter_plot_kwargs(self):
    """
    Returns kwargs dictionary for passing to [matplotlib plot][matplotlib.axes.Axes.plot] method
    """
    return {
        'color': self.color,
        'linewidth': self.size,
        'alpha': self.alpha,
        'zorder': self.zorder,
        'label': self.label,
    }

PlottableData2D pydantic-model

Bases: DataProduct

Base class for children of DataProduct to be plotted ax xy data on a 2D plot

Attributes:

Name	Type	Description
format2d	Format2D | None	Format to apply to plot
tags	Tags	Tags to be used for sorting data.
metadata	dict[str, str]	A dictionary of metadata to be used as a tool tip for mousover in grafana

Show JSON schema:

{
  "$defs": {
    "Format2D": {
      "additionalProperties": false,
      "description": "Formatting data for matplotlib figure and axes\n\nAttributes:\n    title_fig (Optional[str]): Sets [figure title][matplotlib.figure.Figure.suptitle]\n    title_legend (Optional[str]): Sets [legend title][matplotlib.legend.Legend.set_title]\n    title_ax (Optional[str]): Sets [axis title][matplotlib.axes.Axes.set_title]\n    label_x (Optional[str]): Sets [x-axis label][matplotlib.axes.Axes.set_xlabel]\n    label_y (Optional[str]): Sets [y-axis label][matplotlib.axes.Axes.set_ylabel]\n    lim_x_min (float | str | None): Sets [x-axis lower bound][matplotlib.axes.Axes.set_xlim]\n    lim_x_max (float | str | None): Sets [x-axis upper bound][matplotlib.axes.Axes.set_xlim]\n    lim_y_min (float | str | None): Sets [y-axis lower bound][matplotlib.axes.Axes.set_ylim]\n    lim_y_max (float | str | None): Sets [y-axis upper bound][matplotlib.axes.Axes.set_ylim]",
      "properties": {
        "title_fig": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Title Fig"
        },
        "title_legend": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Title Legend"
        },
        "title_ax": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Title Ax"
        },
        "label_x": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Label X"
        },
        "label_y": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Label Y"
        },
        "lim_x_min": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Lim X Min"
        },
        "lim_x_max": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Lim X Max"
        },
        "lim_y_min": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Lim Y Min"
        },
        "lim_y_max": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Lim Y Max"
        }
      },
      "title": "Format2D",
      "type": "object"
    }
  },
  "additionalProperties": true,
  "description": "Base class for children of DataProduct to be plotted ax xy data on a 2D plot\n\nAttributes:\n    format2d (Format2D|None): Format to apply to plot\n    tags (Tags): Tags to be used for sorting data.\n    metadata (dict[str, str]): A dictionary of metadata to be used as a tool tip for mousover in grafana",
  "properties": {
    "tags": {
      "items": {
        "anyOf": []
      },
      "title": "Tags",
      "type": "array"
    },
    "metadata": {
      "additionalProperties": {
        "type": "string"
      },
      "default": {},
      "title": "Metadata",
      "type": "object"
    },
    "format2d": {
      "anyOf": [
        {
          "$ref": "#/$defs/Format2D"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    }
  },
  "required": [
    "tags"
  ],
  "title": "PlottableData2D",
  "type": "object"
}

Fields:

• format2d (Format2D | None)

Point2D pydantic-model

Bases: XYData

Defines a point to be scattered onto xy plot.

Attributes:

Name	Type	Description
tags	Tags	Tags to be used for sorting data.
x	float | str	X value for the point.
y	float | str	Y value for the point.
marker	Marker | None	Style and label information for scattering points to matplotlib axes. Only the label information is used in Grafana. Eventually style information will be used in grafana.
metadata	dict[str, str]	A dictionary of metadata to be used as a tool tip for mousover in grafana

Show JSON schema:

{
  "$defs": {
    "Format2D": {
      "additionalProperties": false,
      "description": "Formatting data for matplotlib figure and axes\n\nAttributes:\n    title_fig (Optional[str]): Sets [figure title][matplotlib.figure.Figure.suptitle]\n    title_legend (Optional[str]): Sets [legend title][matplotlib.legend.Legend.set_title]\n    title_ax (Optional[str]): Sets [axis title][matplotlib.axes.Axes.set_title]\n    label_x (Optional[str]): Sets [x-axis label][matplotlib.axes.Axes.set_xlabel]\n    label_y (Optional[str]): Sets [y-axis label][matplotlib.axes.Axes.set_ylabel]\n    lim_x_min (float | str | None): Sets [x-axis lower bound][matplotlib.axes.Axes.set_xlim]\n    lim_x_max (float | str | None): Sets [x-axis upper bound][matplotlib.axes.Axes.set_xlim]\n    lim_y_min (float | str | None): Sets [y-axis lower bound][matplotlib.axes.Axes.set_ylim]\n    lim_y_max (float | str | None): Sets [y-axis upper bound][matplotlib.axes.Axes.set_ylim]",
      "properties": {
        "title_fig": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Title Fig"
        },
        "title_legend": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Title Legend"
        },
        "title_ax": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Title Ax"
        },
        "label_x": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Label X"
        },
        "label_y": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Label Y"
        },
        "lim_x_min": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Lim X Min"
        },
        "lim_x_max": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Lim X Max"
        },
        "lim_y_min": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Lim Y Min"
        },
        "lim_y_max": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Lim Y Max"
        }
      },
      "title": "Format2D",
      "type": "object"
    },
    "Marker": {
      "additionalProperties": false,
      "description": "Defines marker for scattering to matplotlib\n\nAttributes:\n    color (str): Color of line\n    size (float): Line width\n    alpha (float): Opacity from 0 to 1 (inclusive)\n    zorder (float): Prioritization \n    label (Union[str, None]): Legend label\n    symbol (str): Matplotlib symbol string",
      "properties": {
        "color": {
          "default": "k",
          "title": "Color",
          "type": "string"
        },
        "size": {
          "default": 5,
          "title": "Size",
          "type": "number"
        },
        "alpha": {
          "default": 1,
          "title": "Alpha",
          "type": "number"
        },
        "zorder": {
          "default": 0,
          "title": "Zorder",
          "type": "number"
        },
        "label": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Label"
        },
        "symbol": {
          "default": ".",
          "title": "Symbol",
          "type": "string"
        }
      },
      "title": "Marker",
      "type": "object"
    }
  },
  "additionalProperties": false,
  "description": "Defines a point to be scattered onto xy plot.\n\nAttributes:\n    tags (Tags): Tags to be used for sorting data.        \n    x (float | str): X value for the point.\n    y (float | str): Y value for the point.\n    marker (Marker | None): Style and label information for scattering points to matplotlib axes.\n        Only the label information is used in Grafana.\n        Eventually style information will be used in grafana.\n    metadata (dict[str, str]): A dictionary of metadata to be used as a tool tip for mousover in grafana",
  "properties": {
    "tags": {
      "items": {
        "anyOf": []
      },
      "title": "Tags",
      "type": "array"
    },
    "metadata": {
      "additionalProperties": {
        "type": "string"
      },
      "default": {},
      "title": "Metadata",
      "type": "object"
    },
    "format2d": {
      "anyOf": [
        {
          "$ref": "#/$defs/Format2D"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "x": {
      "anyOf": [
        {
          "type": "number"
        },
        {
          "type": "string"
        }
      ],
      "title": "X"
    },
    "y": {
      "anyOf": [
        {
          "type": "number"
        },
        {
          "type": "string"
        }
      ],
      "title": "Y"
    },
    "marker": {
      "anyOf": [
        {
          "$ref": "#/$defs/Marker"
        },
        {
          "type": "null"
        }
      ],
      "default": {
        "color": "k",
        "size": 5.0,
        "alpha": 1.0,
        "zorder": 0.0,
        "label": null,
        "symbol": "."
      }
    }
  },
  "required": [
    "tags",
    "x",
    "y"
  ],
  "title": "Point2D",
  "type": "object"
}

Config:

• extra: 'forbid'

Fields:

• x (float | str)
• y (float | str)
• marker (Marker | None)

ProductType

Bases: StrEnum

Defines all product types. Used to type-cast URL info in server to validate.

Attributes:

Name	Type	Description
DataProduct	str	class name
XYData	str	class name
Trace2D	str	class name
Point2D	str	class name
TableEntry	str	class name
HistogramEntry	str	class name

SingleAxisFigure dataclass

SingleAxisFigure(tag: Tag, fig: Figure, ax: Axes)

Data class storing a matlab figure and axis. The stored tag data in this class is so-far unused.

Attributes:

Name	Type	Description
ax	Axes	Matplotlib axis to which data will be plotted
fig	Figure	Matplotlib figure.
tag	Tag	Figure tag. Not yet used.

__del__

__del__()

Closes stored matplotlib figure before deleting reference to object.

Source code in src/trendify/API.py

def __del__(self):
    """
    Closes stored matplotlib figure before deleting reference to object.
    """
    plt.close(self.fig)

apply_format

apply_format(format2d: Format2D)

Applies format to figure and axes labels and limits

Parameters:

Name	Type	Description	Default
format2d	Format2D	format information to apply to the single axis figure	required

Source code in src/trendify/API.py

def apply_format(self, format2d: Format2D):
    """
    Applies format to figure and axes labels and limits

    Args:
        format2d (Format2D): format information to apply to the single axis figure
    """
    self.ax.set_title(format2d.title_ax)
    self.fig.suptitle(format2d.title_fig)
    with warnings.catch_warnings(action='ignore', category=UserWarning):
        handles, labels = self.ax.get_legend_handles_labels()
        by_label = dict(zip(labels, handles))
        if by_label:
            self.ax.legend(by_label.values(), by_label.keys(), title=format2d.title_legend)
    self.ax.set_xlabel(format2d.label_x)
    self.ax.set_ylabel(format2d.label_y)
    self.ax.set_xlim(format2d.lim_x_min, format2d.lim_x_max)
    self.ax.set_ylim(format2d.lim_y_min, format2d.lim_y_max)
    self.fig.tight_layout(rect=[0, 0.03, 1, 0.95])
    return self

new classmethod

new(tag: Tag)

Creates new figure and axis. Returns new instance of this class.

Parameters:

Name	Type	Description	Default
tag	Tag	tag (not yet used)	required

Returns:

Type	Description
Type[Self]	New single axis figure

Source code in src/trendify/API.py

@classmethod
def new(cls, tag: Tag):
    """
    Creates new figure and axis.  Returns new instance of this class.

    Args:
        tag (Tag): tag (not yet used)

    Returns:
        (Type[Self]): New single axis figure
    """
    fig: plt.Figure = plt.figure()
    ax: plt.Axes = fig.add_subplot(1, 1, 1)
    return cls(
        tag=tag,
        fig=fig,
        ax=ax,
    )

savefig

savefig(path: Path, dpi: int = 500)

Wrapper on matplotlib savefig method. Saves figure to given path with given dpi resolution.

Returns:

Type	Description
Self	Returns self

Source code in src/trendify/API.py

def savefig(self, path: Path, dpi: int = 500):
    """
    Wrapper on matplotlib savefig method.  Saves figure to given path with given dpi resolution.

    Returns:
        (Self): Returns self
    """
    self.fig.savefig(path, dpi=dpi)
    return self

TableBuilder

TableBuilder(in_dirs: List[Path], out_dir: Path)

Builds tables (melted, pivot, and stats) for histogramming and including in a report or Grafana dashboard.

Parameters:

Name	Type	Description	Default
in_dirs	List[Path]	directories from which to load data products	required
out_dir	Path	directory in which tables should be saved	required

Source code in src/trendify/API.py

def __init__(
        self,
        in_dirs: List[Path],
        out_dir: Path,
    ):
    self.in_dirs = in_dirs
    self.out_dir = out_dir

get_stats_table classmethod

get_stats_table(df: DataFrame)

Computes multiple statistics for each column

Parameters:

Name	Type	Description	Default
df	DataFrame	DataFrame for which the column statistics are to be calculated.	required

Returns:

Type	Description
DataFrame	Dataframe having statistics (column headers) for each of the columns of the input df. The columns of df will be the row indices of the stats table.

Source code in src/trendify/API.py

@classmethod
def get_stats_table(
        cls, 
        df: pd.DataFrame,
    ):
    """
    Computes multiple statistics for each column

    Args:
        df (pd.DataFrame): DataFrame for which the column statistics are to be calculated.

    Returns:
        (pd.DataFrame): Dataframe having statistics (column headers) for each of the columns
            of the input `df`.  The columns of `df` will be the row indices of the stats table.
    """
    # Try to convert to numeric, coerce errors to NaN
    numeric_df = df.apply(pd.to_numeric, errors='coerce')

    stats = {
        'min': numeric_df.min(axis=0),
        'mean': numeric_df.mean(axis=0),
        'max': numeric_df.max(axis=0),
        'sigma3': numeric_df.std(axis=0)*3,
    }
    df_stats = pd.DataFrame(stats, index=df.columns)
    df_stats.index.name = 'Name'
    return df_stats

load_table

load_table(tag: Tag, data_products_fname: str = DATA_PRODUCTS_FNAME_DEFAULT)

Collects table entries from JSON files corresponding to given tag and processes them.

Saves CSV files for the melted data frame, pivot dataframe, and pivot dataframe stats.

File names will all use the tag with different suffixes 'tag_melted.csv', 'tag_pivot.csv', 'name_stats.csv'.

Parameters:

Name	Type	Description	Default
tag	Tag	product tag for which to collect and process.	required

Source code in src/trendify/API.py

def load_table(
        self,
        tag: Tag,
        data_products_fname: str = DATA_PRODUCTS_FNAME_DEFAULT,
    ):
    """
    Collects table entries from JSON files corresponding to given tag and processes them.

    Saves CSV files for the melted data frame, pivot dataframe, and pivot dataframe stats.

    File names will all use the tag with different suffixes 
    `'tag_melted.csv'`, `'tag_pivot.csv'`, `'name_stats.csv'`.

    Args:
        tag (Tag): product tag for which to collect and process.
    """
    print(f'Making table for {tag = }')

    table_entries: List[TableEntry] = []
    for subdir in self.in_dirs:
        collection = DataProductCollection.model_validate_json(subdir.joinpath(data_products_fname).read_text())
        table_entries.extend(collection.get_products(tag=tag, object_type=TableEntry).elements)

    self.process_table_entries(tag=tag, table_entries=table_entries, out_dir=self.out_dir)

process_table_entries classmethod

process_table_entries(tag: Tag, table_entries: List[TableEntry], out_dir: Path)

Saves CSV files for the melted data frame, pivot dataframe, and pivot dataframe stats.

File names will all use the tag with different suffixes 'tag_melted.csv', 'tag_pivot.csv', 'name_stats.csv'.

Parameters:

Name	Type	Description	Default
tag	Tag	product tag for which to collect and process.	required
table_entries	List[TableEntry]	List of table entries	required
out_dir	Path	Directory to which table CSV files should be saved	required

Source code in src/trendify/API.py

@classmethod
def process_table_entries(
        cls,
        tag: Tag,
        table_entries: List[TableEntry],
        out_dir: Path,
    ):
    """

    Saves CSV files for the melted data frame, pivot dataframe, and pivot dataframe stats.

    File names will all use the tag with different suffixes 
    `'tag_melted.csv'`, `'tag_pivot.csv'`, `'name_stats.csv'`.

    Args:
        tag (Tag): product tag for which to collect and process.
        table_entries (List[TableEntry]): List of table entries
        out_dir (Path): Directory to which table CSV files should be saved
    """
    melted = pd.DataFrame([t.get_entry_dict() for t in table_entries])
    pivot = TableEntry.pivot_table(melted=melted)

    save_path_partial = out_dir.joinpath(*tuple(atleast_1d(tag)))
    save_path_partial.parent.mkdir(exist_ok=True, parents=True)
    print(f'Saving to {str(save_path_partial)}_*.csv')

    melted.to_csv(save_path_partial.with_stem(save_path_partial.stem + '_melted').with_suffix('.csv'), index=False)

    if pivot is not None:
        pivot.to_csv(save_path_partial.with_stem(save_path_partial.stem + '_pivot').with_suffix('.csv'), index=True)

        try:
            stats = cls.get_stats_table(df=pivot)
            if not stats.empty and not stats.isna().all().all():
                stats.to_csv(save_path_partial.with_stem(save_path_partial.stem + '_stats').with_suffix('.csv'), index=True)
        except Exception as e:
            print(f'Could not generate pivot table for {tag = }. Error: {str(e)}')

TableEntry pydantic-model

Bases: DataProduct

Defines an entry to be collected into a table.

Collected table entries will be printed in three forms when possible: melted, pivot (when possible), and stats (on pivot columns, when possible).

Attributes:

Name	Type	Description
tags	Tags	Tags used to sort data products
row	float | str	Row Label
col	float | str	Column Label
value	float | str	Value
unit	str | None	Units for value
metadata	dict[str, str]	A dictionary of metadata to be used as a tool tip for mousover in grafana

Show JSON schema:

{
  "additionalProperties": false,
  "description": "Defines an entry to be collected into a table.\n\nCollected table entries will be printed in three forms when possible: melted, pivot (when possible), and stats (on pivot columns, when possible).\n\nAttributes:\n    tags (Tags): Tags used to sort data products\n    row (float | str): Row Label\n    col (float | str): Column Label\n    value (float | str): Value\n    unit (str | None): Units for value\n    metadata (dict[str, str]): A dictionary of metadata to be used as a tool tip for mousover in grafana",
  "properties": {
    "tags": {
      "items": {
        "anyOf": []
      },
      "title": "Tags",
      "type": "array"
    },
    "metadata": {
      "additionalProperties": {
        "type": "string"
      },
      "default": {},
      "title": "Metadata",
      "type": "object"
    },
    "row": {
      "anyOf": [
        {
          "type": "number"
        },
        {
          "type": "string"
        }
      ],
      "title": "Row"
    },
    "col": {
      "anyOf": [
        {
          "type": "number"
        },
        {
          "type": "string"
        }
      ],
      "title": "Col"
    },
    "value": {
      "anyOf": [
        {
          "type": "number"
        },
        {
          "type": "string"
        },
        {
          "type": "boolean"
        }
      ],
      "title": "Value"
    },
    "unit": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Unit"
    }
  },
  "required": [
    "tags",
    "row",
    "col",
    "value"
  ],
  "title": "TableEntry",
  "type": "object"
}

Config:

• extra: 'forbid'

Fields:

• row (float | str)
• col (float | str)
• value (float | str | bool)
• unit (str | None)

get_entry_dict

get_entry_dict()

Returns a dictionary of entries to be used in creating a table.

Returns:

Type	Description
dict[str, str | float]	Dictionary of entries to be used in creating a melted DataFrame

Source code in src/trendify/API.py

def get_entry_dict(self):
    """
    Returns a dictionary of entries to be used in creating a table.

    Returns:
        (dict[str, str | float]): Dictionary of entries to be used in creating a melted [DataFrame][pandas.DataFrame]
    """
    return {'row': self.row, 'col': self.col, 'value': self.value, 'unit': self.unit}

load_and_pivot classmethod

load_and_pivot(path: Path)

Loads melted table from csv and pivots to wide form. csv should have columns named 'row', 'col', and 'value'.

Parameters:

Name	Type	Description	Default
path	Path	path to CSV file	required

Returns:

Type	Description
DataFrame | None	Pivoted data frame or elese None if pivot operation fails.

Source code in src/trendify/API.py

@classmethod
def load_and_pivot(cls, path: Path):
    """
    Loads melted table from csv and pivots to wide form.
    csv should have columns named `'row'`, `'col'`, and `'value'`.

    Args:
        path (Path): path to CSV file

    Returns:
        (pd.DataFrame | None): Pivoted data frame or elese `None` if pivot operation fails.
    """
    return cls.pivot_table(melted=pd.read_csv(path))

pivot_table classmethod

pivot_table(melted: DataFrame)

Attempts to pivot melted row, col, value DataFrame into a wide form DataFrame

Parameters:

Name	Type	Description	Default
melted	DataFrame	Melted data frame having columns named 'row', 'col', 'value'.	required

Returns:

Type	Description
DataFrame | None	pivoted DataFrame if pivot works else None. Pivot operation fails if row or column index pairs are repeated.

Source code in src/trendify/API.py

@classmethod
def pivot_table(cls, melted: pd.DataFrame):
    """
    Attempts to pivot melted row, col, value DataFrame into a wide form DataFrame

    Args:
        melted (pd.DataFrame): Melted data frame having columns named `'row'`, `'col'`, `'value'`.

    Returns:
        (pd.DataFrame | None): pivoted DataFrame if pivot works else `None`. Pivot operation fails if 
            row or column index pairs are repeated.
    """
    try:
        result = melted.pivot(index='row', columns='col', values='value')
    except ValueError as e:
        logger.debug(traceback.format_exc())
        result = None
    return result

Trace2D pydantic-model

Bases: XYData

A collection of points comprising a trace. Use the Trace2D.from_xy constructor.

Attributes:

Name	Type	Description
points	List[Point2D]	List of points. Usually the points would have null values for marker and format2d fields to save space.
pen	Pen	Style and label information for drawing to matplotlib axes. Only the label information is used in Grafana. Eventually style information will be used in grafana.
tags	Tags	Tags to be used for sorting data.
metadata	dict[str, str]	A dictionary of metadata to be used as a tool tip for mousover in grafana

Show JSON schema:

{
  "$defs": {
    "Format2D": {
      "additionalProperties": false,
      "description": "Formatting data for matplotlib figure and axes\n\nAttributes:\n    title_fig (Optional[str]): Sets [figure title][matplotlib.figure.Figure.suptitle]\n    title_legend (Optional[str]): Sets [legend title][matplotlib.legend.Legend.set_title]\n    title_ax (Optional[str]): Sets [axis title][matplotlib.axes.Axes.set_title]\n    label_x (Optional[str]): Sets [x-axis label][matplotlib.axes.Axes.set_xlabel]\n    label_y (Optional[str]): Sets [y-axis label][matplotlib.axes.Axes.set_ylabel]\n    lim_x_min (float | str | None): Sets [x-axis lower bound][matplotlib.axes.Axes.set_xlim]\n    lim_x_max (float | str | None): Sets [x-axis upper bound][matplotlib.axes.Axes.set_xlim]\n    lim_y_min (float | str | None): Sets [y-axis lower bound][matplotlib.axes.Axes.set_ylim]\n    lim_y_max (float | str | None): Sets [y-axis upper bound][matplotlib.axes.Axes.set_ylim]",
      "properties": {
        "title_fig": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Title Fig"
        },
        "title_legend": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Title Legend"
        },
        "title_ax": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Title Ax"
        },
        "label_x": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Label X"
        },
        "label_y": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Label Y"
        },
        "lim_x_min": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Lim X Min"
        },
        "lim_x_max": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Lim X Max"
        },
        "lim_y_min": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Lim Y Min"
        },
        "lim_y_max": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Lim Y Max"
        }
      },
      "title": "Format2D",
      "type": "object"
    },
    "Marker": {
      "additionalProperties": false,
      "description": "Defines marker for scattering to matplotlib\n\nAttributes:\n    color (str): Color of line\n    size (float): Line width\n    alpha (float): Opacity from 0 to 1 (inclusive)\n    zorder (float): Prioritization \n    label (Union[str, None]): Legend label\n    symbol (str): Matplotlib symbol string",
      "properties": {
        "color": {
          "default": "k",
          "title": "Color",
          "type": "string"
        },
        "size": {
          "default": 5,
          "title": "Size",
          "type": "number"
        },
        "alpha": {
          "default": 1,
          "title": "Alpha",
          "type": "number"
        },
        "zorder": {
          "default": 0,
          "title": "Zorder",
          "type": "number"
        },
        "label": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Label"
        },
        "symbol": {
          "default": ".",
          "title": "Symbol",
          "type": "string"
        }
      },
      "title": "Marker",
      "type": "object"
    },
    "Pen": {
      "additionalProperties": false,
      "description": "Defines the pen drawing to matplotlib.\n\nAttributes:\n    color (str): Color of line\n    size (float): Line width\n    alpha (float): Opacity from 0 to 1 (inclusive)\n    zorder (float): Prioritization \n    label (Union[str, None]): Legend label",
      "properties": {
        "color": {
          "default": "k",
          "title": "Color",
          "type": "string"
        },
        "size": {
          "default": 1,
          "title": "Size",
          "type": "number"
        },
        "alpha": {
          "default": 1,
          "title": "Alpha",
          "type": "number"
        },
        "zorder": {
          "default": 0,
          "title": "Zorder",
          "type": "number"
        },
        "label": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Label"
        }
      },
      "title": "Pen",
      "type": "object"
    },
    "Point2D": {
      "additionalProperties": false,
      "description": "Defines a point to be scattered onto xy plot.\n\nAttributes:\n    tags (Tags): Tags to be used for sorting data.        \n    x (float | str): X value for the point.\n    y (float | str): Y value for the point.\n    marker (Marker | None): Style and label information for scattering points to matplotlib axes.\n        Only the label information is used in Grafana.\n        Eventually style information will be used in grafana.\n    metadata (dict[str, str]): A dictionary of metadata to be used as a tool tip for mousover in grafana",
      "properties": {
        "tags": {
          "items": {
            "anyOf": []
          },
          "title": "Tags",
          "type": "array"
        },
        "metadata": {
          "additionalProperties": {
            "type": "string"
          },
          "default": {},
          "title": "Metadata",
          "type": "object"
        },
        "format2d": {
          "anyOf": [
            {
              "$ref": "#/$defs/Format2D"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "x": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "string"
            }
          ],
          "title": "X"
        },
        "y": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "string"
            }
          ],
          "title": "Y"
        },
        "marker": {
          "anyOf": [
            {
              "$ref": "#/$defs/Marker"
            },
            {
              "type": "null"
            }
          ],
          "default": {
            "color": "k",
            "size": 5.0,
            "alpha": 1.0,
            "zorder": 0.0,
            "label": null,
            "symbol": "."
          }
        }
      },
      "required": [
        "tags",
        "x",
        "y"
      ],
      "title": "Point2D",
      "type": "object"
    }
  },
  "additionalProperties": false,
  "description": "A collection of points comprising a trace.\nUse the [Trace2D.from_xy][trendify.API.Trace2D.from_xy] constructor.\n\nAttributes:\n    points (List[Point2D]): List of points.  Usually the points would have null values \n        for `marker` and `format2d` fields to save space.\n    pen (Pen): Style and label information for drawing to matplotlib axes.\n        Only the label information is used in Grafana.\n        Eventually style information will be used in grafana.\n    tags (Tags): Tags to be used for sorting data.\n    metadata (dict[str, str]): A dictionary of metadata to be used as a tool tip for mousover in grafana",
  "properties": {
    "tags": {
      "items": {
        "anyOf": []
      },
      "title": "Tags",
      "type": "array"
    },
    "metadata": {
      "additionalProperties": {
        "type": "string"
      },
      "default": {},
      "title": "Metadata",
      "type": "object"
    },
    "format2d": {
      "anyOf": [
        {
          "$ref": "#/$defs/Format2D"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "points": {
      "items": {
        "$ref": "#/$defs/Point2D"
      },
      "title": "Points",
      "type": "array"
    },
    "pen": {
      "$ref": "#/$defs/Pen",
      "default": {
        "color": "k",
        "size": 1.0,
        "alpha": 1.0,
        "zorder": 0.0,
        "label": null
      }
    }
  },
  "required": [
    "tags",
    "points"
  ],
  "title": "Trace2D",
  "type": "object"
}

Config:

• extra: 'forbid'

Fields:

• points (List[Point2D])
• pen (Pen)
• x (NDArray[Shape['*'], float])
• y (NDArray[Shape['*'], float])

x pydantic-field

x: NDArray[Shape['*'], float]

Returns an array of x values from self.points

Returns:

Type	Description
NDArray[Shape['*'], float]	array of x values from self.points

'

y pydantic-field

y: NDArray[Shape['*'], float]

Returns an array of y values from self.points

Returns:

Type	Description
NDArray[Shape['*'], float]	array of y values from self.points

from_xy classmethod

from_xy(
    tags: Tags,
    x: NDArray[Shape["*"], float],
    y: NDArray[Shape["*"], float],
    pen: Pen = Pen(),
    format2d: Format2D = Format2D(),
)

Creates a list of Point2Ds from xy data and returns a new Trace2D product.

Parameters:

Name	Type	Description	Default
tags	Tags	Tags used to sort data products	required
x	NDArray[Shape['*'], float]	x values	required
y	NDArray[Shape['*'], float]	y values	required
pen	Pen	Style and label for trace	Pen()
format2d	Format2D	Format to apply to plot	Format2D()

Source code in src/trendify/API.py

@classmethod
def from_xy(
        cls,
        tags: Tags,
        x: NDArray[Shape["*"], float],
        y: NDArray[Shape["*"], float],
        pen: Pen = Pen(),
        format2d: Format2D = Format2D(),
    ):
    """
    Creates a list of [Point2D][trendify.API.Point2D]s from xy data and returns a new [Trace2D][trendify.API.Trace2D] product.

    Args:
        tags (Tags): Tags used to sort data products
        x (NDArray[Shape["*"], float]): x values
        y (NDArray[Shape["*"], float]): y values
        pen (Pen): Style and label for trace
        format2d (Format2D): Format to apply to plot
    """
    return cls(
        tags = tags,
        points = [
            Point2D(
                tags=[None],
                x=x_,
                y=y_,
                marker=None,
                format2d=None,
            )
            for x_, y_
            in zip(x, y)
        ],
        pen=pen,
        format2d=format2d,
    )

plot_to_ax

plot_to_ax(ax: Axes)

Plots xy data from trace to a matplotlib axes object.

Parameters:

Name	Type	Description	Default
ax	Axes	axes to which xy data should be plotted	required

Source code in src/trendify/API.py

def plot_to_ax(self, ax: plt.Axes):
    """
    Plots xy data from trace to a matplotlib axes object.

    Args:
        ax (plt.Axes): axes to which xy data should be plotted
    """
    ax.plot(self.x, self.y, **self.pen.as_scatter_plot_kwargs())

propagate_format2d_and_pen

propagate_format2d_and_pen(marker_symbol: str = '.') -> None

Propagates format and style info to all self.points (in-place). I thought this would be useful for grafana before I learned better methods for propagating the data. It still may end up being useful if my plotting method changes. Keeping for potential future use case.

Parameters:

Name	Type	Description	Default
marker_symbol	str	Valid matplotlib marker symbol	'.'

Source code in src/trendify/API.py

def propagate_format2d_and_pen(self, marker_symbol: str = '.') -> None:
    """
    Propagates format and style info to all `self.points` (in-place).
    I thought this would  be useful for grafana before I learned better methods for propagating the data.
    It still may end up being useful if my plotting method changes.  Keeping for potential future use case.

    Args:
        marker_symbol (str): Valid matplotlib marker symbol
    """
    self.points = [
        p.model_copy(
            update={
                'tags': self.tags,
                'format2d': self.format2d,
                'marker': Marker.from_pen(self.pen, symbol=marker_symbol)
            }
        ) 
        for p 
        in self.points
    ]

XYData pydantic-model

Bases: PlottableData2D

Base class for children of DataProduct to be plotted ax xy data on a 2D plot

Show JSON schema:

{
  "$defs": {
    "Format2D": {
      "additionalProperties": false,
      "description": "Formatting data for matplotlib figure and axes\n\nAttributes:\n    title_fig (Optional[str]): Sets [figure title][matplotlib.figure.Figure.suptitle]\n    title_legend (Optional[str]): Sets [legend title][matplotlib.legend.Legend.set_title]\n    title_ax (Optional[str]): Sets [axis title][matplotlib.axes.Axes.set_title]\n    label_x (Optional[str]): Sets [x-axis label][matplotlib.axes.Axes.set_xlabel]\n    label_y (Optional[str]): Sets [y-axis label][matplotlib.axes.Axes.set_ylabel]\n    lim_x_min (float | str | None): Sets [x-axis lower bound][matplotlib.axes.Axes.set_xlim]\n    lim_x_max (float | str | None): Sets [x-axis upper bound][matplotlib.axes.Axes.set_xlim]\n    lim_y_min (float | str | None): Sets [y-axis lower bound][matplotlib.axes.Axes.set_ylim]\n    lim_y_max (float | str | None): Sets [y-axis upper bound][matplotlib.axes.Axes.set_ylim]",
      "properties": {
        "title_fig": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Title Fig"
        },
        "title_legend": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Title Legend"
        },
        "title_ax": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Title Ax"
        },
        "label_x": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Label X"
        },
        "label_y": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Label Y"
        },
        "lim_x_min": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Lim X Min"
        },
        "lim_x_max": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Lim X Max"
        },
        "lim_y_min": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Lim Y Min"
        },
        "lim_y_max": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Lim Y Max"
        }
      },
      "title": "Format2D",
      "type": "object"
    }
  },
  "additionalProperties": true,
  "description": "Base class for children of DataProduct to be plotted ax xy data on a 2D plot",
  "properties": {
    "tags": {
      "items": {
        "anyOf": []
      },
      "title": "Tags",
      "type": "array"
    },
    "metadata": {
      "additionalProperties": {
        "type": "string"
      },
      "default": {},
      "title": "Metadata",
      "type": "object"
    },
    "format2d": {
      "anyOf": [
        {
          "$ref": "#/$defs/Format2D"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    }
  },
  "required": [
    "tags"
  ],
  "title": "XYData",
  "type": "object"
}

XYDataPlotter

XYDataPlotter(in_dirs: List[Path], out_dir: Path, dpi: int = 500)

Plots xy data from user-specified directories to a single axis figure

Parameters:

Name	Type	Description	Default
in_dirs	List[Path]	Directories in which to search for data products from JSON files	required
out_dir	Path	directory to which figure will be output	required
dpi	int	Saved image resolution	500

Source code in src/trendify/API.py

def __init__(
        self,
        in_dirs: List[Path],
        out_dir: Path,
        dpi: int = 500,
    ):
    self.in_dirs = in_dirs
    self.out_dir = out_dir
    self.dpi = dpi

handle_points_and_traces classmethod

handle_points_and_traces(
    tag: Tag,
    points: List[Point2D],
    traces: List[Trace2D],
    axlines: List[AxLine],
    dir_out: Path,
    dpi: int,
)

Plots points, traces, and axlines, formats figure, saves figure, and closes matplotlinb figure.

Parameters:

Name	Type	Description	Default
tag	Tag	Tag corresponding to the provided points and traces	required
points	List[Point2D]	Points to be scattered	required
traces	List[Trace2D]	List of traces to be plotted	required
axlines	List[AxLine]	List of axis lines to be plotted	required
dir_out	Path	directory to output the plot	required
dpi	int	resolution of plot	required

Source code in src/trendify/API.py

@classmethod
def handle_points_and_traces(
        cls,
        tag: Tag,
        points: List[Point2D],
        traces: List[Trace2D],
        axlines: List[AxLine],  # Add this parameter
        dir_out: Path,
        dpi: int,
    ):
    """
    Plots points, traces, and axlines, formats figure, saves figure, and closes matplotlinb figure.

    Args:
        tag (Tag): Tag  corresponding to the provided points and traces
        points (List[Point2D]): Points to be scattered
        traces (List[Trace2D]): List of traces to be plotted
        axlines (List[AxLine]): List of axis lines to be plotted
        dir_out (Path): directory to output the plot
        dpi (int): resolution of plot
    """

    saf = SingleAxisFigure.new(tag=tag)

    if points:
        markers = set([p.marker for p in points])
        for marker in markers:
            matching_points = [p for p in points if p.marker == marker]
            x = [p.x for p in matching_points]
            y = [p.y for p in matching_points]
            if x and y:
                saf.ax.scatter(x, y, **marker.as_scatter_plot_kwargs())

    for trace in traces:
        trace.plot_to_ax(saf.ax)

    # Add plotting of axlines
    for axline in axlines:
        axline.plot_to_ax(saf.ax)

    formats = list(set([p.format2d for p in points] + [t.format2d for t in traces]))
    format2d = Format2D.union_from_iterable(formats)
    saf.apply_format(format2d)
    # saf.ax.autoscale(enable=True, axis='both', tight=True)

    save_path = dir_out.joinpath(*tuple(atleast_1d(tag))).with_suffix('.jpg')
    save_path.parent.mkdir(exist_ok=True, parents=True)
    print(f'Saving to {save_path = }')
    saf.savefig(path=save_path, dpi=dpi)
    del saf

plot

plot(tag: Tag, data_products_fname: str = DATA_PRODUCTS_FNAME_DEFAULT)

• Collects data from json files in stored self.in_dirs,
• plots the relevant products,
• applies labels and formatting,
• saves the figure
• closes matplotlib figure

Parameters:

Name	Type	Description	Default
tag	Tag	data tag for which products are to be collected and plotted.	required
data_products_fname	str	Data products file name	DATA_PRODUCTS_FNAME_DEFAULT

Source code in src/trendify/API.py

def plot(
        self,  
        tag: Tag, 
        data_products_fname: str = DATA_PRODUCTS_FNAME_DEFAULT,
    ):
    """
    - Collects data from json files in stored `self.in_dirs`, 
    - plots the relevant products,
    - applies labels and formatting, 
    - saves the figure
    - closes matplotlib figure

    Args:
        tag (Tag): data tag for which products are to be collected and plotted.
        data_products_fname (str): Data products file name
    """
    print(f'Making xy plot for {tag = }')
    saf = SingleAxisFigure.new(tag=tag)

    for subdir in self.in_dirs:
        collection = DataProductCollection.model_validate_json(subdir.joinpath(data_products_fname).read_text())
        traces: List[Trace2D] = collection.get_products(tag=tag, object_type=Trace2D).elements
        points: List[Point2D] = collection.get_products(tag=tag, object_type=Point2D).elements

        if points or traces:
            if points:
                markers = set([p.marker for p in points])
                for marker in markers:
                    matching_points = [p for p in points if p.marker == marker]
                    x = [p.x for p in matching_points]
                    y = [p.y for p in matching_points]
                    if x and y:
                        if marker is not None:
                            saf.ax.scatter(x, y, **marker.as_scatter_plot_kwargs())
                        else:
                            saf.ax.scatter(x, y)

            for trace in traces:
                trace.plot_to_ax(saf.ax)

            formats = list(set([p.format2d for p in points if p.format2d] + [t.format2d for t in traces]) - {None})
            format2d = Format2D.union_from_iterable(formats)
            saf.apply_format(format2d)
            # saf.ax.autoscale(enable=True, axis='both', tight=True)

    save_path = self.out_dir.joinpath(*tuple(atleast_1d(tag))).with_suffix('.jpg')
    save_path.parent.mkdir(exist_ok=True, parents=True)
    print(f'Saving to {save_path = }')
    saf.savefig(path=save_path, dpi=self.dpi)
    del saf

atleast_1d

atleast_1d(obj: Any) -> Iterable

Converts scalar objec to a list of length 1 or leaves an iterable object unchanged.

Parameters:

Name	Type	Description	Default
obj	Any	Object that needs to be at least 1d	required

Returns:

Type	Description
Iterable	Returns an iterable

Source code in src/trendify/API.py

def atleast_1d(obj: Any) -> Iterable:
    """
    Converts scalar objec to a list of length 1 or leaves an iterable object unchanged.

    Args:
        obj (Any): Object that needs to be at least 1d

    Returns:
        (Iterable): Returns an iterable
    """
    if not should_be_flattened(obj):
        return [obj]
    else:
        return obj

flatten

flatten(obj: Iterable)

Recursively flattens iterable up to a point (leaves str, bytes, and DataProduct unflattened)

Parameters:

Name	Type	Description	Default
obj	Iterable	Object to be flattened	required

Returns:

Type	Description
Iterable	Flattned iterable

Source code in src/trendify/API.py

def flatten(obj: Iterable):
    """
    Recursively flattens iterable up to a point (leaves `str`, `bytes`, and `DataProduct` unflattened)

    Args:
        obj (Iterable): Object to be flattened

    Returns:
        (Iterable): Flattned iterable
    """
    if not should_be_flattened(obj):
        yield obj
    else:
        for sublist in obj:
            yield from flatten(sublist)

get_and_reserve_next_index

get_and_reserve_next_index(save_dir: Path, dir_in: Path)

Reserves next available file index during trendify sorting phase. Saves data to index map file.

Parameters:

Name	Type	Description	Default
save_dir	Path	Directory for which the next available file index is needed	required
dir_in	Path	Directory from which data is being pulled for sorting	required

Source code in src/trendify/API.py

def get_and_reserve_next_index(save_dir: Path, dir_in: Path):
    """
    Reserves next available file index during trendify sorting phase.
    Saves data to index map file.

    Args:
        save_dir (Path): Directory for which the next available file index is needed
        dir_in (Path): Directory from which data is being pulled for sorting
    """
    assert save_dir.is_dir()
    lock_file = save_dir.joinpath('reserving_index.lock')
    with FileLock(lock_file):
        index_map = save_dir.joinpath('index_map.csv')
        index_list = index_map.read_text().strip().split('\n') if index_map.exists() else []
        next_index = int(index_list[-1].split(',')[0])+1 if index_list else 0
        index_list.append(f'{next_index},{dir_in}')
        index_map.write_text('\n'.join(index_list))
    return next_index

get_sorted_dirs

get_sorted_dirs(dirs: List[Path])

Sorts dirs numerically if possible, else alphabetically

Parameters:

Name	Type	Description	Default
dirs	List[Path]	Directories to sort	required

Returns:

Type	Description
List[Path]	Sorted list of directories

Source code in src/trendify/API.py

def get_sorted_dirs(dirs: List[Path]):
    """
    Sorts dirs numerically if possible, else alphabetically

    Args:
        dirs (List[Path]): Directories to sort

    Returns:
        (List[Path]): Sorted list of directories
    """
    dirs = list(dirs)
    try:
        dirs.sort(key=lambda p: int(p.name))
    except ValueError:
        dirs.sort()
    return dirs

make_include_files

make_include_files(
    root_dir: Path,
    local_server_path: str | Path = None,
    mkdocs_include_dir: str | Path = None,
    heading_level: int | None = None,
)

Makes nested include files for inclusion into an MkDocs site.

Note

I recommend to create a Grafana panel and link to that from the MkDocs site instead.

Parameters:

Name	Type	Description	Default
root_dir	Path	Directory for which the include files should be recursively generated	required
local_server_path	str | Path | None	What should the beginning of the path look like? Use //localhost:8001/... something like that to work with python -m mkdocs serve while running python -m http.server 8001 in order to have interactive updates. Use my python convert_links.py script to update after running python -m mkdocs build in order to fix the links for the MkDocs site. See this repo for an example.	None
mkdocs_include_dir	str | Path | None	Path to be used for mkdocs includes. This path should correspond to includ dir in mkdocs.yml file. (See vulcan_srb_sep repo for example).	None

Note:

Here is how to setup `mkdocs.yml` file to have an `include_dir` that can be used to 
include generated markdown files (and the images/CSVs that they reference).

```
plugins:
  - macros:
    include_dir: run_for_record
```

Source code in src/trendify/API.py

def make_include_files(
        root_dir: Path,
        local_server_path: str | Path = None,
        mkdocs_include_dir: str | Path = None,
        # products_dir_replacement_path: str | Path = None,
        heading_level: int | None = None,
    ):
    """
    Makes nested include files for inclusion into an MkDocs site.

    Note:
        I recommend to create a Grafana panel and link to that from the MkDocs site instead.

    Args:
        root_dir (Path): Directory for which the include files should be recursively generated
        local_server_path (str|Path|None): What should the beginning of the path look like?
            Use `//localhost:8001/...` something like that to work with `python -m mkdocs serve`
            while running `python -m http.server 8001` in order to have interactive updates.
            Use my python `convert_links.py` script to update after running `python -m mkdocs build`
            in order to fix the links for the MkDocs site.  See this repo for an example.
        mkdocs_include_dir (str|Path|None): Path to be used for mkdocs includes.
            This path should correspond to includ dir in `mkdocs.yml` file.  (See `vulcan_srb_sep` repo for example).

    Note:

        Here is how to setup `mkdocs.yml` file to have an `include_dir` that can be used to 
        include generated markdown files (and the images/CSVs that they reference).

        ```
        plugins:
          - macros:
            include_dir: run_for_record
        ```

    """

    INCLUDE = 'include.md'
    dirs = list(root_dir.glob('**/'))
    dirs.sort()
    if dirs:
        min_len = np.min([len(list(p.parents)) for p in dirs])
        for s in dirs:
            child_dirs = list(s.glob('*/'))
            child_dirs.sort()
            tables_to_include: List[Path] = [x for x in flatten([list(s.glob(p, case_sensitive=False)) for p in ['*pivot.csv', '*stats.csv']])]
            figures_to_include: List[Path] = [x for x in flatten([list(s.glob(p, case_sensitive=False)) for p in ['*.jpg', '*.png']])]
            children_to_include: List[Path] = [
                c.resolve().joinpath(INCLUDE)
                for c in child_dirs
            ]
            if local_server_path is not None:
                figures_to_include = [
                    Path(local_server_path).joinpath(x.relative_to(root_dir))
                    for x in figures_to_include
                ]
            if mkdocs_include_dir is not None:
                tables_to_include = [
                    x.relative_to(mkdocs_include_dir.parent)
                    for x in tables_to_include
                ]
                children_to_include = [
                    x.relative_to(mkdocs_include_dir)
                    for x in children_to_include
                ]

            bb_open = r'{{'
            bb_close = r'}}'
            fig_inclusion_statements = [
                f'![]({x})' 
                for x in figures_to_include
            ]
            table_inclusion_statements = [
                f"{bb_open} read_csv('{x}', disable_numparse=True) {bb_close}"
                for x in tables_to_include
            ]
            child_inclusion_statments = [
                "{% include '" + str(x) + "' %}"
                for x in children_to_include
            ]
            fig_inclusion_statements.sort()
            table_inclusion_statements.sort()
            child_inclusion_statments.sort()
            inclusions = table_inclusion_statements + fig_inclusion_statements + child_inclusion_statments

            header = (
                ''.join(['#']*((len(list(s.parents))-min_len)+heading_level)) + s.name 
                if heading_level is not None and len(inclusions) > 1
                else ''
            )
            text = '\n\n'.join([header] + inclusions)

            s.joinpath(INCLUDE).write_text(text)

make_it_trendy

make_it_trendy(
    data_product_generator: ProductGenerator | None,
    input_dirs: List[Path],
    output_dir: Path,
    n_procs: int = 1,
    dpi_static_plots: int = 500,
    no_static_tables: bool = False,
    no_static_xy_plots: bool = False,
    no_static_histograms: bool = False,
    no_grafana_dashboard: bool = False,
    no_include_files: bool = False,
    protocol: str = "http",
    server: str = "0.0.0.0",
    port: int = 8000,
    data_products_fname: str = DATA_PRODUCTS_FNAME_DEFAULT,
)

Maps data_product_generator over dirs_in to produce data product JSON files in those directories. Sorts the generated data products into a nested file structure starting from dir_products. Nested folders are generated for tags that are Tuples. Sorted data files are named according to the directory from which they were loaded.

Parameters:

Name	Type	Description	Default
data_product_generator	ProductGenerator | None	A callable function that returns a list of data products given a working directory.	required
input_dirs	List[Path]	Directories over which to map the product_generator	required
output_dir	Path	Directory to which the trendify products and assets will be written.	required
n_procs	int = 1	Number of processes to run in parallel. If n_procs==1, directories will be processed sequentially (easier for debugging since the full traceback will be provided). If n_procs > 1, a ProcessPoolExecutor will be used to load and process directories and/or tags in parallel.	1
dpi_static_plots	int = 500	Resolution of output plots when using matplotlib (for make_xy_plots==True and/or make_histograms==True)	500
no_static_tables	bool	Suppresses static assets from the TableEntry products	False
no_static_xy_plots	bool	Suppresses static assets from the XYData (Trace2D and Point2D) products	False
no_static_histograms	bool	Suppresses static assets from the HistogramEntry products	False
no_grafana_dashboard	bool	Suppresses generation of Grafana dashboard JSON definition file	False
no_include_files	bool	Suppresses generation of include files for importing static assets to markdown or LaTeX reports	False
data_products_fname	str	File name to be used for storing generated data products	DATA_PRODUCTS_FNAME_DEFAULT

Source code in src/trendify/API.py

def make_it_trendy(
        data_product_generator: ProductGenerator | None,
        input_dirs: List[Path],
        output_dir: Path,
        n_procs: int = 1,
        dpi_static_plots: int = 500,
        no_static_tables: bool = False,
        no_static_xy_plots: bool = False,
        no_static_histograms: bool = False,
        no_grafana_dashboard: bool = False,
        no_include_files: bool = False,
        protocol: str = 'http',
        server: str = '0.0.0.0',
        port: int = 8000,
        data_products_fname: str = DATA_PRODUCTS_FNAME_DEFAULT,
    ):
    """
    Maps `data_product_generator` over `dirs_in` to produce data product JSON files in those directories.
    Sorts the generated data products into a nested file structure starting from `dir_products`.
    Nested folders are generated for tags that are Tuples.  Sorted data files are named according to the
    directory from which they were loaded.

    Args:
        data_product_generator (ProductGenerator | None): A callable function that returns
            a list of data products given a working directory.
        input_dirs (List[Path]): Directories over which to map the `product_generator`
        output_dir (Path): Directory to which the trendify products and assets will be written.
        n_procs (int = 1): Number of processes to run in parallel.  If `n_procs==1`, directories will be
            processed sequentially (easier for debugging since the full traceback will be provided).
            If `n_procs > 1`, a [ProcessPoolExecutor][concurrent.futures.ProcessPoolExecutor] will
            be used to load and process directories and/or tags in parallel.
        dpi_static_plots (int = 500): Resolution of output plots when using matplotlib 
            (for `make_xy_plots==True` and/or `make_histograms==True`)
        no_static_tables (bool): Suppresses static assets from the [`TableEntry`][trendify.API.TableEntry] products
        no_static_xy_plots (bool): Suppresses static assets from the 
            [`XYData`][trendify.API.XYData] 
            ([Trace2D][trendify.API.Trace2D] and [Point2D][trendify.API.Point2D]) products
        no_static_histograms (bool): Suppresses static assets from the [`HistogramEntry`][trendify.API.HistogramEntry] products
        no_grafana_dashboard (bool): Suppresses generation of Grafana dashboard JSON definition file
        no_include_files (bool): Suppresses generation of include files for importing static assets to markdown or LaTeX reports
        data_products_fname (str): File name to be used for storing generated data products
    """
    input_dirs = [Path(p).parent if Path(p).is_file() else Path(p) for p in list(input_dirs)]
    output_dir = Path(output_dir)

    make_products(
        product_generator=data_product_generator,
        data_dirs=input_dirs,
        n_procs=n_procs,
        data_products_fname=data_products_fname,
    )

    products_dir = _mkdir(output_dir.joinpath('products'))

    # Sort products
    start = time.time()
    sort_products(
        data_dirs=input_dirs,
        output_dir=products_dir,
        n_procs=n_procs,
        data_products_fname=data_products_fname,
    )
    end = time.time()
    print(f'Time to sort = {end - start}')

    no_static_assets = (no_static_tables and no_static_histograms and no_static_xy_plots)
    no_interactive_assets = (no_grafana_dashboard)
    no_assets = no_static_assets and no_interactive_assets

    if not no_assets:
        assets_dir = output_dir.joinpath('assets')
        if not no_interactive_assets:
            interactive_assets_dir = _mkdir(assets_dir.joinpath('interactive'))
            if not no_grafana_dashboard:
                grafana_dir = _mkdir(interactive_assets_dir.joinpath('grafana'))
                make_grafana_dashboard(
                    products_dir=products_dir,
                    output_dir=grafana_dir,
                    n_procs=n_procs,
                    protocol=protocol,
                    server=server,
                    port=port,
                )

        if not no_static_assets:
            static_assets_dir = _mkdir(assets_dir.joinpath('static'))
            make_tables_and_figures(
                products_dir=products_dir,
                output_dir=static_assets_dir,
                dpi=dpi_static_plots,
                n_procs=n_procs,
                no_tables=no_static_tables,
                no_xy_plots=no_static_xy_plots,
                no_histograms=no_static_histograms,
            )

            if not no_include_files:
                make_include_files(
                    root_dir=static_assets_dir,
                    heading_level=2,
                )

make_products

make_products(
    product_generator: Callable[[Path], DataProductCollection] | None,
    data_dirs: List[Path],
    n_procs: int = 1,
    data_products_fname: str = DATA_PRODUCTS_FNAME_DEFAULT,
)

Maps product_generator over dirs_in to produce data product JSON files in those directories. Sorts the generated data products into a nested file structure starting from dir_products. Nested folders are generated for tags that are Tuples. Sorted data files are named according to the directory from which they were loaded.

Parameters:

Name	Type	Description	Default
product_generator	ProductGenerator | None	A callable function that returns a list of data products given a working directory.	required
data_dirs	List[Path]	Directories over which to map the product_generator	required
n_procs	int = 1	Number of processes to run in parallel. If n_procs==1, directories will be processed sequentially (easier for debugging since the full traceback will be provided). If n_procs > 1, a ProcessPoolExecutor will be used to load and process directories and/or tags in parallel.	1
data_products_fname	str	File name to be used for storing generated data products	DATA_PRODUCTS_FNAME_DEFAULT

Source code in src/trendify/API.py

def make_products(
        product_generator: Callable[[Path], DataProductCollection] | None,
        data_dirs: List[Path],
        n_procs: int = 1,
        data_products_fname: str = DATA_PRODUCTS_FNAME_DEFAULT,
    ):
    """
    Maps `product_generator` over `dirs_in` to produce data product JSON files in those directories.
    Sorts the generated data products into a nested file structure starting from `dir_products`.
    Nested folders are generated for tags that are Tuples.  Sorted data files are named according to the
    directory from which they were loaded.

    Args:
        product_generator (ProductGenerator | None): A callable function that returns
            a list of data products given a working directory.
        data_dirs (List[Path]): Directories over which to map the `product_generator`
        n_procs (int = 1): Number of processes to run in parallel.  If `n_procs==1`, directories will be
            processed sequentially (easier for debugging since the full traceback will be provided).
            If `n_procs > 1`, a [ProcessPoolExecutor][concurrent.futures.ProcessPoolExecutor] will
            be used to load and process directories and/or tags in parallel.
        data_products_fname (str): File name to be used for storing generated data products
    """
    sorted_dirs = get_sorted_dirs(dirs=data_dirs)

    if product_generator is None:
        print('No data product generator provided')
    else:
        print('\n\n\nGenerating tagged DataProducts and writing to JSON files...\n')
        map_callable(
            DataProductGenerator(processor=product_generator).process_and_save,
            sorted_dirs,
            [data_products_fname]*len(sorted_dirs),
            n_procs=n_procs,
        )
        print('\nFinished generating tagged DataProducts and writing to JSON files')

make_tables_and_figures

make_tables_and_figures(
    products_dir: Path,
    output_dir: Path,
    dpi: int = 500,
    n_procs: int = 1,
    no_tables: bool = False,
    no_xy_plots: bool = False,
    no_histograms: bool = False,
)

Makes CSV tables and creates plots (using matplotlib).

Tags will be processed in parallel and output in nested directory structure under output_dir.

Parameters:

Name	Type	Description	Default
products_dir	Path	Directory to which the sorted data products will be written	required
output_dir	Path	Directory to which tables and matplotlib histograms and plots will be written if the appropriate boolean variables make_tables, make_xy_plots, make_histograms are true.	required
n_procs	int = 1	Number of processes to run in parallel. If n_procs==1, directories will be processed sequentially (easier for debugging since the full traceback will be provided). If n_procs > 1, a ProcessPoolExecutor will be used to load and process directories and/or tags in parallel.	1
dpi	int = 500	Resolution of output plots when using matplotlib (for make_xy_plots==True and/or make_histograms==True)	500
no_tables	bool	Whether or not to collect the TableEntry products and write them to CSV files (<tag>_melted.csv with <tag>_pivot.csv and <tag>_stats.csv when possible).	False
no_xy_plots	bool	Whether or not to plot the XYData products using matplotlib	False
no_histograms	bool	Whether or not to generate histograms of the HistogramEntry products using matplotlib.	False

Source code in src/trendify/API.py

def make_tables_and_figures(
        products_dir: Path,
        output_dir: Path,
        dpi: int = 500,
        n_procs: int = 1,
        no_tables: bool = False,
        no_xy_plots: bool = False,
        no_histograms: bool = False,
    ):
    """
    Makes CSV tables and creates plots (using matplotlib).

    Tags will be processed in parallel and output in nested directory structure under `output_dir`.

    Args:
        products_dir (Path): Directory to which the sorted data products will be written
        output_dir (Path): Directory to which tables and matplotlib histograms and plots will be written if
            the appropriate boolean variables `make_tables`, `make_xy_plots`, `make_histograms` are true.
        n_procs (int = 1): Number of processes to run in parallel.  If `n_procs==1`, directories will be
            processed sequentially (easier for debugging since the full traceback will be provided).
            If `n_procs > 1`, a [ProcessPoolExecutor][concurrent.futures.ProcessPoolExecutor] will
            be used to load and process directories and/or tags in parallel.
        dpi (int = 500): Resolution of output plots when using matplotlib 
            (for `make_xy_plots==True` and/or `make_histograms==True`)
        no_tables (bool): Whether or not to collect the 
            [`TableEntry`][trendify.API.TableEntry] products and write them
            to CSV files (`<tag>_melted.csv` with `<tag>_pivot.csv` and `<tag>_stats.csv` when possible).
        no_xy_plots (bool): Whether or not to plot the [`XYData`][trendify.API.XYData] products using matplotlib
        no_histograms (bool): Whether or not to generate histograms of the 
            [`HistogramEntry`][trendify.API.HistogramEntry] products
            using matplotlib.
    """
    if not (no_tables and no_xy_plots and no_histograms):
        product_dirs = list(products_dir.glob('**/*/'))
        map_callable(
            DataProductCollection.process_collection,
            product_dirs,
            [output_dir]*len(product_dirs),
            [no_tables]*len(product_dirs),
            [no_xy_plots]*len(product_dirs),
            [no_histograms]*len(product_dirs),
            [dpi]*len(product_dirs),
            n_procs=n_procs,
        )

map_callable

map_callable(
    f: Callable[[Path], DataProductCollection], *iterables, n_procs: int = 1, mp_context=None
)

Parameters:

Name	Type	Description	Default
f	Callable[[Path], DataProductCollection]	Function to be mapped	required
iterables	Tuple[Iterable, ...]	iterables of arguments for mapped function f	()
n_procs	int	Number of parallel processes to run	1
mp_context	str	Context to use for creating new processes (see multiprocessing package documentation)	None

Source code in src/trendify/API.py

def map_callable(
        f: Callable[[Path], DataProductCollection], 
        *iterables, 
        n_procs: int=1, 
        mp_context=None,
    ):
    """
    Args:
        f (Callable[[Path], DataProductCollection]): Function to be mapped
        iterables (Tuple[Iterable, ...]): iterables of arguments for mapped function `f`
        n_procs (int): Number of parallel processes to run
        mp_context (str): Context to use for creating new processes (see `multiprocessing` package documentation)
    """
    if n_procs > 1:
        with ProcessPoolExecutor(max_workers=n_procs, mp_context=mp_context) as executor:
            result = list(executor.map(f, *iterables))
    else:
        result = [f(*arg_tuple) for arg_tuple in zip(*iterables)]

    return result

serve_products_to_plotly_dashboard

serve_products_to_plotly_dashboard(
    *dirs: Path,
    title: str = "Trendify Autodash",
    host: str = "127.0.0.1",
    port: int = 8000,
    debug: bool = False,
    data_products_filename: str = DATA_PRODUCTS_FNAME_DEFAULT
)

Source code in src/trendify/API.py

def serve_products_to_plotly_dashboard(
        *dirs: Path,
        title: str = 'Trendify Autodash',
        host: str = '127.0.0.1',
        port: int = 8000,
        debug: bool = False,
        data_products_filename: str = DATA_PRODUCTS_FNAME_DEFAULT,
    ):
    """
    """
    collection = DataProductCollection.collect_from_all_jsons(*dirs, data_products_filename=data_products_filename)
    collection.serve_plotly_dashboard(
        title=title,
        debug=debug, 
        host=host, 
        port=port,
    )

should_be_flattened

should_be_flattened(obj: Any)

Checks if object is an iterable container that should be flattened. DataProducts will not be flattened. Strings will not be flattened. Everything else will be flattened.

Parameters:

Name	Type	Description	Default
obj	Any	Object to be tested	required

Returns:

Type	Description
bool	Whether or not to flatten object

Source code in src/trendify/API.py

def should_be_flattened(obj: Any):
    """
    Checks if object is an iterable container that should be flattened.
    `DataProduct`s will not be flattened.  Strings will not be flattened.
    Everything else will be flattened.

    Args:
        obj (Any): Object to be tested

    Returns:
        (bool): Whether or not to flatten object
    """
    return isinstance(obj, Iterable) and not isinstance(obj, (str, bytes, DataProduct))

sort_products

sort_products(
    data_dirs: List[Path],
    output_dir: Path,
    n_procs: int = 1,
    data_products_fname: str = DATA_PRODUCTS_FNAME_DEFAULT,
)

Loads the tagged data products from data_dirs and sorts them (by tag) into a nested folder structure rooted at output_dir.

Parameters:

Name	Type	Description	Default
data_dirs	List[Path]	Directories containing JSON data product files	required
output_dir	Path	Directory to which sorted products will be written	required
data_products_fname	str	File name in which the data products to be sorted are stored	DATA_PRODUCTS_FNAME_DEFAULT

Source code in src/trendify/API.py

def sort_products(
        data_dirs: List[Path],
        output_dir: Path,
        n_procs: int = 1,
        data_products_fname: str = DATA_PRODUCTS_FNAME_DEFAULT,
    ):
    """
    Loads the tagged data products from `data_dirs` and sorts them (by tag) into a nested folder structure rooted at `output_dir`.

    Args:
        data_dirs (List[Path]): Directories containing JSON data product files
        output_dir (Path): Directory to which sorted products will be written
        data_products_fname (str): File name in which the data products to be sorted are stored
    """
    sorted_data_dirs = get_sorted_dirs(dirs=data_dirs)

    print('\n\n\nSorting data by tags')
    output_dir.mkdir(parents=True, exist_ok=True)

    map_callable(
        DataProductCollection.sort_by_tags_single_directory,
        sorted_data_dirs,
        [output_dir]*len(sorted_data_dirs),
        [data_products_fname]*len(sorted_data_dirs),
        n_procs=n_procs,
    )

    print('\nFinished sorting by tags')

squeeze

squeeze(obj: Union[Iterable, Any])

Returns a scalar if object is iterable of length 1 else returns object.

Parameters:

Name	Type	Description	Default
obj	Union[Iterable, Any]	An object to be squeezed if possible	required

Returns:

Type	Description
Any	Either iterable or scalar if possible

Source code in src/trendify/API.py

def squeeze(obj: Union[Iterable, Any]):
    """
    Returns a scalar if object is iterable of length 1 else returns object.

    Args:
        obj (Union[Iterable, Any]): An object to be squeezed if possible

    Returns:
        (Any): Either iterable or scalar if possible
    """
    if should_be_flattened(obj) and len(obj) == 1:
        return obj[0]
    else:
        return obj