Add synthesis and diagrams in PNG format

requirements.txt

@@ -4,6 +4,8 @@ coverage~=7.6.10
 flake8~=7.1.1
 interrogate~=1.7.0
 mypy~=1.14.1
+openai~=1.63.0
+python-dotenv~=1.0.1
 Sphinx~=8.1.3
 sphinx-rtd-theme~=3.0.2
 twine~=6.1.0

setup.py

@@ -42,6 +42,9 @@
         'Tracker': 'https://github.com/blueskysolarracing/stpa/issues',
     },
     packages=find_packages(),
+    install_requires=[
+        'openai>=1.63.0,<2',
+    ],
     python_requires='>=3.11',
     package_data={'stpa': ['py.typed']},
 )

stpa/__init__.py

@@ -8,10 +8,19 @@
     'ControlType',
     'Definition',
     'Entity',
+    'EQUALITY_QUERY_MODEL',
+    'EQUALITY_QUERY_PROMPT',
+    'generate_raw_unsafe_control_actions',
     'GEOMETRY_TAG_NAME',
     'Hazard',
     'HTML_PARSER',
+    'LINKAGE_QUERY_MODEL',
+    'LINKAGE_QUERY_PROMPT',
     'Loss',
+    'query_equality',
+    'query_linkage',
+    'RAW_UNSAFE_CONTROL_ACTIONS_GENERATION_MODEL',
+    'RAW_UNSAFE_CONTROL_ACTIONS_GENERATION_PROMPT',
     'Responsibility',
     'Scenario',
     'ScenarioType1',
@@ -22,6 +31,7 @@
     'SystemLevelConstraintType2',
     'SystemLevelConstraintType3',
     'UnsafeControlAction',
+    'YesOrNoResponse',
 )
 
 from stpa.definitions import (
@@ -49,4 +59,15 @@
     Entity,
     GEOMETRY_TAG_NAME,
 )
-from stpa.utilities import clean_html_text, HTML_PARSER
+from stpa.utilities import clean_html_text, HTML_PARSER, YesOrNoResponse
+from stpa.synthesis import (
+    EQUALITY_QUERY_MODEL,
+    EQUALITY_QUERY_PROMPT,
+    generate_raw_unsafe_control_actions,
+    LINKAGE_QUERY_MODEL,
+    LINKAGE_QUERY_PROMPT,
+    query_equality,
+    query_linkage,
+    RAW_UNSAFE_CONTROL_ACTIONS_GENERATION_MODEL,
+    RAW_UNSAFE_CONTROL_ACTIONS_GENERATION_PROMPT,
+)

stpa/synthesis.py

@@ -0,0 +1,192 @@
+from base64 import b64encode
+from collections.abc import Iterable, Iterator
+from re import compile, findall, MULTILINE, sub
+
+from openai import OpenAI
+
+from stpa.definitions import Definition, UnsafeControlAction
+from stpa.utilities import YesOrNoResponse
+
+
+EQUALITY_QUERY_MODEL = 'gpt-4o'
+EQUALITY_QUERY_PROMPT = '''
+Are the STPA Definitions {} and {} saying similar things?
+
+{}
+{}
+
+Answer only "YES" or "NO".
+'''.strip()
+
+
+def query_equality(
+        client: OpenAI,
+        definition_0: Definition,
+        definition_1: Definition,
+        seed: int | None = None,
+) -> bool:
+    prompt = EQUALITY_QUERY_PROMPT.format(
+        definition_0.name,
+        definition_1.name,
+        definition_0,
+        definition_1,
+    )
+    completion = client.chat.completions.create(
+        model=EQUALITY_QUERY_MODEL,
+        messages=[{'role': 'user', 'content': prompt}],
+        seed=seed,
+    )
+    response = completion.choices[0].message.content
+
+    match response:
+        case YesOrNoResponse.YES:
+            status = True
+        case YesOrNoResponse.NO:
+            status = False
+        case _:
+            raise ValueError(r'cannot determine verdict of {repr(response)}')
+
+    return status
+
+
+_DEFINITION_LINKS_PATTERN = compile(r'\s+\[.*?\]$')
+
+
+def _get_definition_name_body(definition: Definition) -> str:
+    return sub(_DEFINITION_LINKS_PATTERN, '', str(definition))
+
+
+_DEFINITION_NAME_PATTERN = compile(r'^.+?:\s+')
+
+
+def _get_definition_body(definition: Definition) -> str:
+    body = _get_definition_name_body(definition)
+
+    return sub(_DEFINITION_NAME_PATTERN, '', body)
+
+
+def _number_lines(lines: Iterable[str]) -> Iterator[str]:
+    for i, line in enumerate(lines):
+        yield f'{i + 1}. {line}'
+
+
+_NUMBER_PATTERN = compile(r'^\d+\. ')
+
+
+def _unnumber_lines(lines: Iterable[str]) -> Iterator[str]:
+    for line in lines:
+        yield sub(_NUMBER_PATTERN, '', line)
+
+
+RAW_UNSAFE_CONTROL_ACTIONS_GENERATION_MODEL = 'gpt-4o'
+RAW_UNSAFE_CONTROL_ACTIONS_GENERATION_PROMPT = '''
+Generate {} unsafe control actions of the system shown in the control\
+ structure diagram.
+
+An unsafe control action follows one of the four patterns: 1) not providing\
+ causes hazard 2) providing causes hazard 3) too early, too late, out of order\
+ or 4) stopped too soon, applied too long.
+
+Examples:
+
+{}
+
+Write each unsafe control action in a single line, preceded by a number.
+'''.strip()
+RAW_UNSAFE_CONTROL_ACTION_PATTERN = compile(r'^\d+\. .+$', MULTILINE)
+
+
+def generate_raw_unsafe_control_actions(
+        client: OpenAI,
+        count: int,
+        example_unsafe_control_actions: Iterable[UnsafeControlAction],
+        diagram_pathname: str,
+        seed: int | None = None,
+) -> list[str]:
+    example_lines = _number_lines(
+        map(_get_definition_body, example_unsafe_control_actions),
+    )
+    prompt = RAW_UNSAFE_CONTROL_ACTIONS_GENERATION_PROMPT.format(
+        count,
+        '\n'.join(example_lines),
+    )
+
+    with open(diagram_pathname, 'rb') as file:
+        base64_image = b64encode(file.read()).decode('utf-8')
+
+    completion = client.chat.completions.create(
+        model=RAW_UNSAFE_CONTROL_ACTIONS_GENERATION_MODEL,
+        messages=[
+            {
+                'role': 'user',
+                'content': [
+                    {
+                        'type': 'text',
+                        'text': prompt,
+                    },
+                    {
+                        'type': 'image_url',
+                        'image_url': {
+                            'url': f'data:image/jpeg;base64,{base64_image}',
+                        },
+                    },
+                ],
+            },
+        ],
+        seed=seed,
+    )
+    response = completion.choices[0].message.content
+
+    assert isinstance(response, str)
+
+    raw_unsafe_control_actions = list(
+        _unnumber_lines(
+            map(
+                str.strip,
+                findall(RAW_UNSAFE_CONTROL_ACTION_PATTERN, response),
+            ),
+        ),
+    )
+
+    return raw_unsafe_control_actions
+
+
+LINKAGE_QUERY_MODEL = 'gpt-4o'
+LINKAGE_QUERY_PROMPT = '''
+Should the STPA Definition {} be linked to Definition {}?
+
+{}
+{}
+
+Answer only "YES" or "NO".
+'''.strip()
+
+
+def query_linkage(
+        client: OpenAI,
+        linked_definition: Definition,
+        linking_definition: Definition,
+        seed: int | None = None,
+) -> bool:
+    prompt = LINKAGE_QUERY_PROMPT.format(
+        linked_definition.name,
+        linking_definition.name,
+        _get_definition_name_body(linked_definition),
+        _get_definition_name_body(linking_definition),
+    )
+    completion = client.chat.completions.create(
+        model=LINKAGE_QUERY_MODEL,
+        messages=[{'role': 'user', 'content': prompt}],
+        seed=seed,
+    )
+    response = completion.choices[0].message.content
+
+    match response:
+        case YesOrNoResponse.YES:
+            status = True
+        case YesOrNoResponse.NO:
+            status = False
+        case _:
+            raise ValueError(r'cannot determine verdict of {repr(response)}')
+
+    return status

stpa/utilities.py

@@ -1,3 +1,5 @@
+from enum import StrEnum
+
 from bs4 import BeautifulSoup
 
 HTML_PARSER = 'html.parser'
@@ -7,3 +9,8 @@ def clean_html_text(html_text: str) -> str:
     soup = BeautifulSoup(html_text, HTML_PARSER)
 
     return soup.get_text(strip=True)
+
+
+class YesOrNoResponse(StrEnum):
+    YES = 'YES'
+    NO = 'NO'