Skip to content

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
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
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
65
66
67
68
69
70
71
72
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
74
75
76
77
78
79
80
81
82
83
84
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
86
87
88
89
90
91
92
93
94
95
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
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
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
120
121
122
123
124
125
126
127
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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
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
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
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(*args, **kwargs)

Report creating an entity link. Not currently supported.

Raises:

Type Description
NotImplementedError
Source code in arkindex_worker/reporting.py
171
172
173
174
175
176
177
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
179
180
181
182
183
184
185
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
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
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
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
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
228
229
230
231
232
233
234
235
236
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)