Skip to content

Commit 764da8d

Browse files
committed
add generic filters
1 parent 0ec5217 commit 764da8d

2 files changed

Lines changed: 205 additions & 87 deletions

File tree

src/extensions/score_metamodel/checks/standards.py

Lines changed: 17 additions & 87 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,13 @@
1414

1515
from sphinx_needs.need_item import NeedItem
1616

17+
from ..sphinx_filters import (
18+
generic_pie_items_by_tag,
19+
generic_pie_items_in_relationships,
20+
generic_pie_linked_items,
21+
generic_pie_workproducts_by_type,
22+
)
23+
1724
# from score_metamodel import (
1825
# CheckLogger,
1926
# graph_check,
@@ -186,27 +193,8 @@ def my_pie_linked_standard_requirements(
186193
Passed arguments can be accessed via kwargs['arg<position>']
187194
See: https://sphinx-needs.readthedocs.io/en/latest/filter.html#arguments
188195
"""
189-
cnt_connected = 0
190-
cnt_not_connected = 0
191-
192-
standard = kwargs["arg1"]
193-
194-
all_standards_needs = get_standards_needs(needs)
195-
standards_needs = {
196-
k: v
197-
for k, v in all_standards_needs.items()
198-
if k.startswith(f"std_req__{standard}__")
199-
}
200-
compliance_req_needs = get_compliance_req_needs(needs)
201-
202-
for need in standards_needs.values():
203-
if need["id"] in compliance_req_needs:
204-
cnt_connected += 1
205-
else:
206-
cnt_not_connected += 1
207-
208-
results.append(cnt_connected)
209-
results.append(cnt_not_connected)
196+
standard = str(kwargs["arg1"])
197+
generic_pie_linked_items(needs, results, arg1=f"std_req__{standard}__", arg2="gd_")
210198

211199

212200
def my_pie_linked_standard_requirements_by_tag(
@@ -238,23 +226,10 @@ def my_pie_linked_standard_requirements_by_tag(
238226
the mutated `results`list, and use this to display/generate the piechart.
239227
240228
"""
241-
count_linked = 0
242-
count_non_linked = 0
243-
244-
tag = str(kwargs["arg1"])
245229
assert len(kwargs) == 1, (
246230
"Can only provide one tag to `my_pie_linked_standard_requirements_by_tag`"
247231
)
248-
249-
compliance_req_needs = get_compliance_req_needs(needs)
250-
for need in needs:
251-
if tag in need["tags"]:
252-
if need["id"] in compliance_req_needs:
253-
count_linked += 1
254-
else:
255-
count_non_linked += 1
256-
results.append(count_linked)
257-
results.append(count_non_linked)
232+
generic_pie_items_by_tag(needs, results, arg1=kwargs["arg1"], arg2="gd_")
258233

259234

260235
def my_pie_linked_standard_workproducts(
@@ -267,28 +242,10 @@ def my_pie_linked_standard_workproducts(
267242
Passed arguments can be accessed via kwargs['arg<position>']
268243
See: https://sphinx-needs.readthedocs.io/en/latest/filter.html#arguments
269244
"""
270-
cwp_connected = 0
271-
cwp_not_connected = 0
272-
273-
standard = kwargs["arg1"]
274-
275-
all_standard_workproducts = get_standards_workproducts(needs)
276-
standard_workproducts = {
277-
k: v
278-
for k, v in all_standard_workproducts.items()
279-
if k.startswith(f"std_wp__{standard}__")
280-
}
281-
282-
compliance_wp_needs = get_compliance_wp_needs(needs)
283-
284-
for need in standard_workproducts.values():
285-
if need["id"] in compliance_wp_needs:
286-
cwp_connected += 1
287-
else:
288-
cwp_not_connected += 1
289-
290-
results.append(cwp_connected)
291-
results.append(cwp_not_connected)
245+
standard = str(kwargs["arg1"])
246+
generic_pie_workproducts_by_type(
247+
needs, results, arg1=f"std_wp__{standard}__", arg2="workproduct"
248+
)
292249

293250

294251
def my_pie_workproducts_contained_in_exactly_one_workflow(
@@ -299,33 +256,6 @@ def my_pie_workproducts_contained_in_exactly_one_workflow(
299256
in exactly one workflow, the not connected once and the once
300257
that are connected to multiple workflows.
301258
"""
302-
all_workflows = get_workflows(needs)
303-
all_workproducts = get_workproducts(needs)
304-
305-
# Map to track counts for each workproduct and their associated workflows
306-
workproduct_analysis = {wp["id"]: {"count": 0} for wp in all_workproducts.values()}
307-
308-
# Iterate over workflows and update the counts and workflows
309-
for workflow in all_workflows.values():
310-
for output in workflow.get("output", []):
311-
# Increment count and add workflow_id if workproduct is in analysis
312-
if output in workproduct_analysis:
313-
workproduct_analysis[output]["count"] += 1
314-
315-
not_connected_wp = 0
316-
nb_wp_connected_to_one_workflow = 0
317-
nb_wp_connected_to_more_than_one_workflow = 0
318-
319-
for analysis in workproduct_analysis.values():
320-
count = analysis["count"]
321-
322-
if count == 0:
323-
not_connected_wp += 1
324-
elif count == 1:
325-
nb_wp_connected_to_one_workflow += 1
326-
else:
327-
nb_wp_connected_to_more_than_one_workflow += 1
328-
329-
results.append(not_connected_wp)
330-
results.append(nb_wp_connected_to_one_workflow)
331-
results.append(nb_wp_connected_to_more_than_one_workflow)
259+
generic_pie_items_in_relationships(
260+
needs, results, arg1="workflow", arg2="output", arg3="workproduct"
261+
)
Lines changed: 188 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,188 @@
1+
# *******************************************************************************
2+
# Copyright (c) 2025 Contributors to the Eclipse Foundation
3+
#
4+
# See the NOTICE file(s) distributed with this work for additional
5+
# information regarding copyright ownership.
6+
#
7+
# This program and the accompanying materials are made available under the
8+
# terms of the Apache License Version 2.0 which is available at
9+
# https://www.apache.org/licenses/LICENSE-2.0
10+
#
11+
# SPDX-License-Identifier: Apache-2.0
12+
# *******************************************************************************
13+
14+
"""Generic sphinx-needs filter functions for ``needpie`` directives.
15+
16+
These functions are fully parameterizable and designed to be called directly
17+
by consumers of docs-as-code (e.g. reference-integration repos) when they
18+
pull in the ``score_docs_as_code`` Bazel module. All functions follow the
19+
sphinx-needs ``filter-func`` signature convention:
20+
21+
.. code-block:: python
22+
23+
def func(needs: list[NeedItem], results: list[int], **kwargs) -> None: ...
24+
25+
Arguments are injected from the ``:filter-func:`` call-site as positional
26+
``arg1``, ``arg2``, … keyword arguments.
27+
28+
Example usage in RST::
29+
30+
.. needpie:: My Requirements Coverage
31+
:labels: Linked, Not Linked
32+
:filter-func: score_metamodel.sphinx_filters.generic_pie_linked_items(std_req__mystandard__, gd_)
33+
34+
"""
35+
36+
from __future__ import annotations
37+
38+
from sphinx_needs.need_item import NeedItem
39+
40+
41+
def generic_pie_linked_items(
42+
needs: list[NeedItem], results: list[int], **kwargs: str | int | float
43+
) -> None:
44+
"""Count items matching an ID prefix split by compliance linkage.
45+
46+
Finds all needs whose ``id`` starts with *arg1*, then checks whether
47+
each one appears in the ``complies`` field of any need whose ``type``
48+
starts with *arg2*.
49+
50+
:filter-func: arguments:
51+
52+
- ``arg1`` – ID prefix of the items to count
53+
(e.g. ``std_req__iso26262__``)
54+
- ``arg2`` – type prefix of the source needs whose ``complies``
55+
lists are scanned (e.g. ``gd_``)
56+
57+
Appends to *results*: ``[linked_count, not_linked_count]``
58+
"""
59+
id_prefix = str(kwargs.get("arg1", ""))
60+
compliance_prefix = str(kwargs.get("arg2", ""))
61+
62+
target_ids = [
63+
str(n.get("id", ""))
64+
for n in needs
65+
if str(n.get("id", "")).startswith(id_prefix)
66+
]
67+
68+
linked_ids: set[str] = {
69+
ref
70+
for n in needs
71+
if str(n.get("type", "")).startswith(compliance_prefix)
72+
for ref in n.get("complies", [])
73+
if ref
74+
}
75+
76+
connected = sum(1 for item_id in target_ids if item_id in linked_ids)
77+
not_connected = len(target_ids) - connected
78+
79+
results.append(connected)
80+
results.append(not_connected)
81+
82+
83+
def generic_pie_items_by_tag(
84+
needs: list[NeedItem], results: list[int], **kwargs: str | int | float
85+
) -> None:
86+
"""Count items carrying a given tag split by compliance linkage.
87+
88+
Checks every need that has *arg1* in its ``tags`` field and splits them
89+
by whether their id appears in the ``complies`` field of any need whose
90+
``type`` starts with *arg2*.
91+
92+
:filter-func: arguments:
93+
94+
- ``arg1`` – tag to filter by (e.g. ``aspice40_man5``).
95+
Note: tag values must not contain dots.
96+
- ``arg2`` – type prefix of the source needs whose ``complies``
97+
lists are scanned (e.g. ``gd_``)
98+
99+
Appends to *results*: ``[linked_count, not_linked_count]``
100+
"""
101+
tag = str(kwargs.get("arg1", ""))
102+
compliance_prefix = str(kwargs.get("arg2", ""))
103+
104+
linked_ids: set[str] = {
105+
ref
106+
for n in needs
107+
if str(n.get("type", "")).startswith(compliance_prefix)
108+
for ref in n.get("complies", [])
109+
if ref
110+
}
111+
112+
linked = 0
113+
not_linked = 0
114+
for n in needs:
115+
if tag in n.get("tags", []):
116+
if str(n.get("id", "")) in linked_ids:
117+
linked += 1
118+
else:
119+
not_linked += 1
120+
121+
results.append(linked)
122+
results.append(not_linked)
123+
124+
125+
def generic_pie_workproducts_by_type(
126+
needs: list[NeedItem], results: list[int], **kwargs: str | int | float
127+
) -> None:
128+
"""Count work-product items matching an ID prefix split by compliance linkage.
129+
130+
Semantically equivalent to :func:`generic_pie_linked_items` but scoped to
131+
work-product traceability where the compliance source type is typically an
132+
exact match (e.g. ``workproduct``) rather than a prefix. Because
133+
``"workproduct".startswith("workproduct")`` is ``True``, both functions use
134+
the same underlying logic.
135+
136+
:filter-func: arguments:
137+
138+
- ``arg1`` – ID prefix of the work-product items to count
139+
(e.g. ``std_wp__iso26262__``)
140+
- ``arg2`` – type (or type prefix) of source needs whose ``complies``
141+
lists are scanned (e.g. ``workproduct``)
142+
143+
Appends to *results*: ``[linked_count, not_linked_count]``
144+
"""
145+
generic_pie_linked_items(needs, results, **kwargs)
146+
147+
148+
def generic_pie_items_in_relationships(
149+
needs: list[NeedItem], results: list[int], **kwargs: str | int | float
150+
) -> None:
151+
"""Count items of a given type by how many container items reference them.
152+
153+
For every need of type *arg3*, counts how many needs of type *arg1*
154+
include its id in their *arg2* field. Splits the result into three
155+
buckets: not referenced, referenced exactly once, referenced more than
156+
once.
157+
158+
:filter-func: arguments:
159+
160+
- ``arg1`` – type of the container needs (e.g. ``workflow``)
161+
- ``arg2`` – field on the container that holds references
162+
(e.g. ``output``)
163+
- ``arg3`` – type of the items to count (e.g. ``workproduct``)
164+
165+
Appends to *results*:
166+
``[not_referenced_count, referenced_once_count, referenced_multiple_count]``
167+
"""
168+
container_type = str(kwargs.get("arg1", ""))
169+
field = str(kwargs.get("arg2", ""))
170+
item_type = str(kwargs.get("arg3", ""))
171+
172+
containers = [n for n in needs if n.get("type") == container_type]
173+
items = [n for n in needs if n.get("type") == item_type]
174+
175+
item_counts: dict[str, int] = {str(n.get("id", "")): 0 for n in items}
176+
177+
for container in containers:
178+
for ref in container.get(field, []):
179+
if ref in item_counts:
180+
item_counts[ref] += 1
181+
182+
not_referenced = sum(1 for c in item_counts.values() if c == 0)
183+
referenced_once = sum(1 for c in item_counts.values() if c == 1)
184+
referenced_multiple = sum(1 for c in item_counts.values() if c > 1)
185+
186+
results.append(not_referenced)
187+
results.append(referenced_once)
188+
results.append(referenced_multiple)

0 commit comments

Comments
 (0)