Reporting

arkindex_worker.reporting

Generator for the ml_report.json file, to report created worker results and exceptions.

Classes

Reporter

Reporter(
    name="Unknown worker",
    slug="unknown-slug",
    version=None,
    **kwargs
)

Bases: object

Helper to generate an ml_report.json artifact.

Source code in arkindex_worker/reporting.py

def __init__(
    self,
    name: Optional[str] = "Unknown worker",
    slug: Optional[str] = "unknown-slug",
    version: Optional[str] = None,
    **kwargs,
):
    self.report_data = {
        "name": name,
        "slug": slug,
        "version": version,
        "started": datetime.utcnow().isoformat(),
        "elements": {},
    }
    logger.info(f"Starting ML report for {name}")

Functions

process

process(element_id)

Report that a specific element ID is being processed.

Parameters:

Name	Type	Description	Default
element_id	Union[str, UUID]	ID of the element being processed.	required

Source code in arkindex_worker/reporting.py

def process(self, element_id: Union[str, UUID]):
    """
    Report that a specific element ID is being processed.

    :param element_id: ID of the element being processed.
    """
    # Just call the element initializer
    self._get_element(element_id)

add_element

add_element(parent_id, type, type_count=1)

Report creating an element as a child of another.

Parameters:

Name	Type	Description	Default
parent_id	Union[str, UUID]	ID of the parent element.	required
type	str	Slug of the type of the child element.	required
type_count	int	How many elements of this type were created.	1

Source code in arkindex_worker/reporting.py

def add_element(self, parent_id: Union[str, UUID], type: str, type_count: int = 1):
    """
    Report creating an element as a child of another.

    :param parent_id: ID of the parent element.
    :param type: Slug of the type of the child element.
    :param type_count: How many elements of this type were created.
    """
    elements = self._get_element(parent_id)["elements"]
    elements.setdefault(type, 0)
    elements[type] += type_count

add_classification

add_classification(element_id, class_name)

Report creating a classification on an element.

Parameters:

Name	Type	Description	Default
element_id	Union[str, UUID]	ID of the element.	required
class_name	str	Name of the ML class of the new classification.	required

Source code in arkindex_worker/reporting.py

def add_classification(self, element_id: Union[str, UUID], class_name: str):
    """
    Report creating a classification on an element.

    :param element_id: ID of the element.
    :param class_name: Name of the ML class of the new classification.
    """
    classifications = self._get_element(element_id)["classifications"]
    classifications.setdefault(class_name, 0)
    classifications[class_name] += 1

add_classifications

add_classifications(element_id, classifications)

Report creating one or more classifications at once on an element.

Parameters:

Name	Type	Description	Default
element_id	Union[str, UUID]	ID of the element.	required
classifications	List[Dict[str, str]]	List of classifications. Each classification is represented as a dict with a class_name key holding the name of the ML class being used.	required

Source code in arkindex_worker/reporting.py

def add_classifications(
    self, element_id: Union[str, UUID], classifications: List[Dict[str, str]]
):
    """
    Report creating one or more classifications at once on an element.

    :param element_id: ID of the element.
    :param classifications: List of classifications.
       Each classification is represented as a ``dict`` with a ``class_name`` key
       holding the name of the ML class being used.
    """
    assert isinstance(
        classifications, list
    ), "A list is required for classifications"
    element = self._get_element(element_id)
    # Retrieve the previous existing classification counts, if any
    counter = Counter(**element["classifications"])
    # Add the new ones
    counter.update(
        [classification["class_name"] for classification in classifications]
    )
    element["classifications"] = dict(counter)

add_transcription

add_transcription(element_id, count=1)

Report creating a transcription on an element.

Parameters:

Name	Type	Description	Default
element_id	Union[str, UUID]	ID of the element.	required
count		Number of transcriptions created at once	1

Source code in arkindex_worker/reporting.py

def add_transcription(self, element_id: Union[str, UUID], count=1):
    """
    Report creating a transcription on an element.

    :param element_id: ID of the element.
    :param count: Number of transcriptions created at once
    """
    self._get_element(element_id)["transcriptions"] += count

add_entity

add_entity(element_id, entity_id, type, name)

Report creating an entity on an element.

Parameters:

Name	Type	Description	Default
element_id	Union[str, UUID]	ID of the element.	required
entity_id	Union[str, UUID]	ID of the new entity.	required
type	str	Type of the entity.	required
name	str	Name of the entity.	required

Source code in arkindex_worker/reporting.py

def add_entity(
    self,
    element_id: Union[str, UUID],
    entity_id: Union[str, UUID],
    type: str,
    name: str,
):
    """
    Report creating an entity on an element.

    :param element_id: ID of the element.
    :param entity_id: ID of the new entity.
    :param type: Type of the entity.
    :param name: Name of the entity.
    """
    entities = self._get_element(element_id)["entities"]
    entities.append({"id": entity_id, "type": type, "name": name})

add_transcription_entity

add_transcription_entity(
    entity_id, transcription, transcription_entity_id
)

Report creating a transcription entity on an element.

Parameters:

Name	Type	Description	Default
entity_id	Union[str, UUID]	ID of the entity element.	required
transcription	Transcription	Transcription to add the entity on	required
transcription_entity_id	Union[str, UUID]	ID of the transcription entity that is created.	required

Source code in arkindex_worker/reporting.py

def add_transcription_entity(
    self,
    entity_id: Union[str, UUID],
    transcription: Transcription,
    transcription_entity_id: Union[str, UUID],
):
    """
    Report creating a transcription entity on an element.

    :param entity_id: ID of the entity element.
    :param transcription: Transcription to add the entity on
    :param transcription_entity_id: ID of the transcription entity that is created.
    """
    transcription_entities = self._get_element(transcription.element.id)[
        "transcription_entities"
    ]
    transcription_entities.append(
        {
            "transcription_id": transcription.id,
            "entity_id": entity_id,
            "transcription_entity_id": transcription_entity_id,
        }
    )

add_entity_link

add_entity_link(*args, **kwargs)

Report creating an entity link. Not currently supported.

Raises:

Type	Description
NotImplementedError

Source code in arkindex_worker/reporting.py

def add_entity_link(self, *args, **kwargs):
    """
    Report creating an entity link. Not currently supported.

    :raises NotImplementedError:
    """
    raise NotImplementedError

add_entity_role

add_entity_role(*args, **kwargs)

Report creating an entity role. Not currently supported.

Raises:

Type	Description
NotImplementedError

Source code in arkindex_worker/reporting.py

def add_entity_role(self, *args, **kwargs):
    """
    Report creating an entity role. Not currently supported.

    :raises NotImplementedError:
    """
    raise NotImplementedError

add_metadata

add_metadata(element_id, metadata_id, type, name)

Report creating a metadata from an element.

Parameters:

Name	Type	Description	Default
element_id	Union[str, UUID]	ID of the element.	required
metadata_id	Union[str, UUID]	ID of the new metadata.	required
type	str	Type of the metadata.	required
name	str	Name of the metadata.	required

Source code in arkindex_worker/reporting.py

def add_metadata(
    self,
    element_id: Union[str, UUID],
    metadata_id: Union[str, UUID],
    type: str,
    name: str,
):
    """
    Report creating a metadata from an element.

    :param element_id: ID of the element.
    :param metadata_id: ID of the new metadata.
    :param type: Type of the metadata.
    :param name: Name of the metadata.
    """
    metadata = self._get_element(element_id)["metadata"]
    metadata.append({"id": metadata_id, "type": type, "name": name})

error

error(element_id, exception)

Report that a Python exception occurred when processing an element.

Parameters:

Name	Type	Description	Default
element_id	Union[str, UUID]	ID of the element.	required
exception	Exception	A Python exception.	required

Source code in arkindex_worker/reporting.py

def error(self, element_id: Union[str, UUID], exception: Exception):
    """
    Report that a Python exception occurred when processing an element.

    :param element_id: ID of the element.
    :param exception: A Python exception.
    """
    error_data = {
        "class": exception.__class__.__name__,
        "message": str(exception),
    }
    if exception.__traceback__ is not None:
        error_data["traceback"] = "\n".join(
            traceback.format_tb(exception.__traceback__)
        )

    if isinstance(exception, ErrorResponse):
        error_data["message"] = exception.title
        error_data["status_code"] = exception.status_code
        error_data["content"] = exception.content

    self._get_element(element_id)["errors"].append(error_data)

save

save(path)

Save the ML report to the specified path.

Parameters:

Name	Type	Description	Default
path	Union[str, Path]	Path to save the ML report to.	required

Source code in arkindex_worker/reporting.py

def save(self, path: Union[str, Path]):
    """
    Save the ML report to the specified path.

    :param path: Path to save the ML report to.
    """
    logger.info(f"Saving ML report to {path}")
    with open(path, "w") as f:
        json.dump(self.report_data, f)