analytics.py

"""BHoM analytics decorator."""
# pylint: disable=E0401
import codecs
from dataclasses import dataclass, field
import inspect
from itertools import groupby
import itertools
import json
import os
from pathlib import Path
import socket
import sys
import traceback
import uuid
from functools import wraps
from typing import Any, Callable, Dict, List, Union
from datetime import datetime

# pylint: enable=E0401

from .logging import ANALYTICS_LOGGER, CONSOLE_LOGGER
from .util import bson_unix_ticks, bson_unix_ticks_to_datetime
from . import BHOM_VERSION, TOOLKIT_NAME, BHOM_LOG_FOLDER, DISABLE_ANALYTICS

@dataclass
class UsageLogEntry():
    BHoMVersion:str = BHOM_VERSION
    BHoM_Guid:uuid.UUID = uuid.uuid4()
    CallerName:str = ""
    ComponentId:uuid.UUID = uuid.uuid4()
    CustomData:Dict = field(default_factory = {"interpreter", sys.executable})
    Errors:List[str] = field(default_factory = [])
    FileId:str = ""
    FileName:str = ""
    Fragments:List[str] = field(default_factory = [])
    Name:str = ""
    ProjectID:str = ""
    SelectedItem:Dict = field(default_factory = {"MethodName": "", "Parameters": [], "TypeName": ""})
    Time:Dict = field(default_factory = {"$date": 0})
    UI:str = "Python"
    UiVersion:str = sys.version
    _t:str = "BH.oM.Base.UsageLogEntry"

    @classmethod
    def from_json(cls, json_str:str) -> 'UsageLogEntry':
        d = json.loads(json_str)
        if "CustomData" not in d:
            d["CustomData"] = None
        if "Fragments" not in d:
            d["Fragments"] = None
        return UsageLogEntry(d["BHoMVersion"], d["BHoM_Guid"], d["CallerName"], d["ComponentId"], d["CustomData"], d["Errors"], d["FileId"], d["FileName"], d["Fragments"], d["Name"], d["ProjectID"], d["SelectedItem"], d["Time"], d["UI"], d["UiVersion"])

def load_logs_from_file(filename:str) -> List[UsageLogEntry]:
    logs:List[UsageLogEntry] = []

    #adapted from https://stackoverflow.com/questions/30629297/remove-byte-order-mark-from-objects-in-a-list
    #due to some files generated by BHoM logs being encoded with utf-8 BOM instead of utf-8
    with open(filename, "r") as f:
        lines = f.readlines()
        if lines[0].__contains__(codecs.BOM_UTF8.decode(f.encoding)):
            # A Byte Order Mark is present
            lines[0] = lines[0].strip(codecs.BOM_UTF8.decode(f.encoding))
        for line in lines:
            if len(line) != 0:
                logs.append(UsageLogEntry.from_json(line))

    return logs

def summarise_usage_logs(usage_log_entries:List[UsageLogEntry]) -> List[Dict]:
    db_entries:List[Dict] = []

    usage_log_entries.sort(key=lambda x: x.ProjectID)

    for file_id, filegroup in groupby(usage_log_entries, lambda x: x.FileId):
        filegroup = list(filegroup)
        project_id = filegroup[0].ProjectID
        filename = filegroup[0].FileName
        filegroup.sort(key = lambda x: x.CallerName + str(x.SelectedItem))

        for method_name, methodgroup in groupby(filegroup, lambda x: x.CallerName + str(x.SelectedItem)):
            methodgroup = list(methodgroup)
            first_entry = methodgroup[0]

            db_entries.append({
                "StartTime": bson_unix_ticks_to_datetime(min(methodgroup, key=lambda x: x.Time["$date"]).Time["$date"], short=True),
                "EndTime": bson_unix_ticks_to_datetime(max(methodgroup, key=lambda x: x.Time["$date"]).Time["$date"], short=True),
                "UI": first_entry.UI,
                "UiVersion":first_entry.UiVersion,
                "CallerName": first_entry.CallerName,
                "SelectedItem": first_entry.SelectedItem,
                "Computer": socket.gethostname(),
                "UserName": os.environ.get("USERNAME"),
                "BHoMVersion": BHOM_VERSION,
                "FileId": file_id,
                "FileName": filename,
                "ProjectID": project_id,
                "NbCallingComponents": len(set([a.ComponentId for a in methodgroup])),
                "TotalNbCals": len(methodgroup),
                "Errors": list(itertools.chain.from_iterable([x.Errors for x in methodgroup])),
                "_t": "BH.oM.BHoMAnalytics.UsageEntry"
            })

    return db_entries

def convert_exc_info_to_bhom_error(exc_info):
    time = bson_unix_ticks(datetime.now(), short=True)
    utcTime = bson_unix_ticks(short=True)
    stack_trace = traceback.extract_tb(exc_info[2])
    message = str(exc_info[1])
    Type = "Error" #using string but ideally this would be an enum value.
    return {"Time": {"$date": time}, "UtcTime": {"$date": utcTime}, "StackTrace": stack_trace, "Message": message, "Type": Type, "_t": "BH.oM.Base.Debugging.Event"}

global PROJECT_NUMBER
PROJECT_NUMBER = None

def set_project_number(project_number: Union[str, None]):
    global PROJECT_NUMBER
    CONSOLE_LOGGER.debug(f"Setting project number: {PROJECT_NUMBER} to {project_number}")
    PROJECT_NUMBER = project_number
    
def get_project_number() -> Union[str, None]:
    CONSOLE_LOGGER.debug(f"Retrieving project number: {PROJECT_NUMBER}")
    return PROJECT_NUMBER

def bhom_analytics(project_id:Callable = get_project_number, disable:bool = DISABLE_ANALYTICS) -> Callable:
    """Decorator for capturing usage data.

    Returns
    -------
    Callable
        The decorated function.
    """

    _componentId = uuid.uuid4()

    def decorator(function: Callable):
        """A decorator to capture usage data for called methods/functions.

        Arguments
        ---------
        function : Callable
            The function to decorate.

        Returns
        -------
        Callable
            The decorated function.
        """

        @wraps(function)
        def wrapper(*args, **kwargs) -> Any:
            """A wrapper around the function that captures usage analytics."""

            if disable:
                CONSOLE_LOGGER.debug("bhom_analytics is curently disabled.")
                return function(*args, **kwargs)
            
            _id = uuid.uuid4()

            #for now for file IDs, generate one using the project ID
            pid = project_id()
            if pid == None:
                pid = ""

            file_id = uuid.uuid3(uuid.NAMESPACE_OID, pid)

            # get the data being passed to the function, expected dtype and return type
            argspec = inspect.getfullargspec(function)[-1]
            argspec.pop("return", None)

            _args = [f'{{"_t": "{argspec[k]}", "Name": "{k}"}}' for k in argspec.keys()]

            exec_metadata = {
                "BHoMVersion": BHOM_VERSION,
                "BHoM_Guid": _id,
                "CallerName": function.__name__,
                "ComponentId": _componentId,
                "CustomData": {"interpreter": sys.executable},
                "Errors": [],
                "FileId": str(file_id),
                "FileName": str(file_id),
                "Fragments": [],
                "Name": "",
                "ProjectID": project_id(),
                "SelectedItem": {
                    "MethodName": function.__name__,
                    "Parameters": _args,
                    "TypeName": f"{function.__module__}.{function.__qualname__}"
                },
                "Time": {
                    "$date": bson_unix_ticks(short=True),
                },
                "UI": "Python",
                "UiVersion": sys.version,
                "_t": "BH.oM.Base.UsageLogEntry",
            }

            try:
                result = function(*args, **kwargs)
            except Exception as exc:  # pylint: disable=broad-except
                exec_metadata["Errors"].extend(convert_exc_info_to_bhom_error(sys.exc_info()))
                raise exc
            finally:
                try:
                    log_file = BHOM_LOG_FOLDER / f"Usage_{function.__module__.split('.')[0]}_{datetime.now().strftime('%Y%m%d')}.log"

                    if ANALYTICS_LOGGER.handlers[0].baseFilename != str(log_file):
                        ANALYTICS_LOGGER.handlers[0].close()
                        ANALYTICS_LOGGER.handlers[0].baseFilename = str(log_file)

                    ANALYTICS_LOGGER.info(
                        json.dumps(exec_metadata, default=str, indent=None)
                    )
                except Exception:
                    pass

            return result

        return wrapper

    return decorator