Skip to content

Pattern

Architectural pattern representation.

Overview

The Pattern class represents reusable architectural patterns with proven success rates.

Class Documentation

upir.patterns.pattern.Pattern dataclass

An architectural pattern discovered from or applied to UPIRs.

Patterns represent common architectural structures that appear across multiple UPIR instances. They can be discovered through clustering or defined manually as templates.

A pattern includes a template structure, metadata about instances that match the pattern, and performance metrics for pattern effectiveness.

Attributes:

Name Type Description
id str

Unique identifier for the pattern
name str

Human-readable name (e.g., "streaming-etl", "api-gateway")
description str

Detailed description of the pattern
template Dict[str, Any]

Template structure with parameterizable components Format: { "components": [{"type": "...", "properties": {...}}], "connections": [{"from": "...", "to": "...", ...}], "parameters": {...} # Tunable parameters }
instances List[str]

List of UPIR IDs that match this pattern
success_rate float

Fraction of instances meeting their specifications (0-1)
average_performance Dict[str, float]

Average metrics across instances Format: {"latency_p99": ..., "throughput_qps": ...}
Example

pattern = Pattern( ... id="streaming-etl-1", ... name="Streaming ETL Pipeline", ... description="Event-driven data processing pipeline", ... template={ ... "components": [ ... {"type": "pubsub_source", "properties": {}}, ... {"type": "stream_processor", "properties": {}}, ... {"type": "bigquery_sink", "properties": {}} ... ], ... "parameters": {"window_size": 60, "parallelism": 10} ... }, ... instances=["upir-1", "upir-2", "upir-3"], ... success_rate=0.95, ... average_performance={"latency_p99": 150, "throughput_qps": 5000} ... )

References: - TD Commons: Pattern extraction and reuse - Design patterns: Template method pattern

Source code in upir/patterns/pattern.py 
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
@dataclass
class Pattern:
    """
    An architectural pattern discovered from or applied to UPIRs.

    Patterns represent common architectural structures that appear across
    multiple UPIR instances. They can be discovered through clustering or
    defined manually as templates.

    A pattern includes a template structure, metadata about instances that
    match the pattern, and performance metrics for pattern effectiveness.

    Attributes:
        id: Unique identifier for the pattern
        name: Human-readable name (e.g., "streaming-etl", "api-gateway")
        description: Detailed description of the pattern
        template: Template structure with parameterizable components
                 Format: {
                     "components": [{"type": "...", "properties": {...}}],
                     "connections": [{"from": "...", "to": "...", ...}],
                     "parameters": {...}  # Tunable parameters
                 }
        instances: List of UPIR IDs that match this pattern
        success_rate: Fraction of instances meeting their specifications (0-1)
        average_performance: Average metrics across instances
                            Format: {"latency_p99": ..., "throughput_qps": ...}

    Example:
        >>> pattern = Pattern(
        ...     id="streaming-etl-1",
        ...     name="Streaming ETL Pipeline",
        ...     description="Event-driven data processing pipeline",
        ...     template={
        ...         "components": [
        ...             {"type": "pubsub_source", "properties": {}},
        ...             {"type": "stream_processor", "properties": {}},
        ...             {"type": "bigquery_sink", "properties": {}}
        ...         ],
        ...         "parameters": {"window_size": 60, "parallelism": 10}
        ...     },
        ...     instances=["upir-1", "upir-2", "upir-3"],
        ...     success_rate=0.95,
        ...     average_performance={"latency_p99": 150, "throughput_qps": 5000}
        ... )

    References:
    - TD Commons: Pattern extraction and reuse
    - Design patterns: Template method pattern
    """

    id: str
    name: str
    description: str
    template: Dict[str, Any]
    instances: List[str] = field(default_factory=list)
    success_rate: float = 0.0
    average_performance: Dict[str, float] = field(default_factory=dict)

    def __post_init__(self):
        """Validate pattern fields."""
        if not self.id:
            raise ValueError("Pattern id cannot be empty")
        if not self.name:
            raise ValueError("Pattern name cannot be empty")
        if not 0.0 <= self.success_rate <= 1.0:
            raise ValueError(f"Success rate must be in [0, 1], got {self.success_rate}")

    def add_instance(self, upir_id: str, performance: Dict[str, float] = None):
        """
        Add a UPIR instance to this pattern.

        Updates instances list and recalculates average performance.

        Args:
            upir_id: ID of the UPIR instance
            performance: Performance metrics for this instance
        """
        if upir_id not in self.instances:
            self.instances.append(upir_id)

        # Update average performance if metrics provided
        if performance:
            for metric, value in performance.items():
                if metric not in self.average_performance:
                    self.average_performance[metric] = value
                else:
                    # Incremental average: new_avg = old_avg + (new_val - old_avg) / count
                    count = len(self.instances)
                    old_avg = self.average_performance[metric]
                    self.average_performance[metric] = old_avg + (value - old_avg) / count

    def matches(self, feature_vector: List[float], threshold: float = 0.8) -> bool:
        """
        Check if a feature vector matches this pattern.

        Uses cosine similarity between the feature vector and the pattern's
        centroid (if available in template).

        Args:
            feature_vector: Feature vector to check
            threshold: Similarity threshold (0-1)

        Returns:
            True if similarity >= threshold
        """
        if "centroid" not in self.template:
            return False

        import numpy as np
        centroid = np.array(self.template["centroid"])
        vector = np.array(feature_vector)

        # Cosine similarity
        dot_product = np.dot(centroid, vector)
        norm_product = np.linalg.norm(centroid) * np.linalg.norm(vector)

        if norm_product == 0:
            return False

        similarity = dot_product / norm_product
        return similarity >= threshold

    def to_dict(self) -> Dict[str, Any]:
        """
        Serialize pattern to dictionary.

        Returns:
            Dictionary representation
        """
        return {
            "id": self.id,
            "name": self.name,
            "description": self.description,
            "template": self.template,
            "instances": self.instances,
            "success_rate": self.success_rate,
            "average_performance": self.average_performance,
        }

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> "Pattern":
        """
        Deserialize pattern from dictionary.

        Args:
            data: Dictionary containing pattern fields

        Returns:
            Pattern instance
        """
        return cls(
            id=data["id"],
            name=data["name"],
            description=data["description"],
            template=data.get("template", {}),
            instances=data.get("instances", []),
            success_rate=data.get("success_rate", 0.0),
            average_performance=data.get("average_performance", {}),
        )

    def __str__(self) -> str:
        """Human-readable representation."""
        return (
            f"Pattern({self.name}, "
            f"{len(self.instances)} instances, "
            f"success_rate={self.success_rate:.2f})"
        )

    def __repr__(self) -> str:
        """Developer-friendly representation."""
        return (
            f"Pattern(id='{self.id}', name='{self.name}', "
            f"instances={len(self.instances)}, "
            f"success_rate={self.success_rate:.2f})"
        )

Functions

__post_init__()

Validate pattern fields.

Source code in upir/patterns/pattern.py 
77
78
79
80
81
82
83
84
def __post_init__(self):
    """Validate pattern fields."""
    if not self.id:
        raise ValueError("Pattern id cannot be empty")
    if not self.name:
        raise ValueError("Pattern name cannot be empty")
    if not 0.0 <= self.success_rate <= 1.0:
        raise ValueError(f"Success rate must be in [0, 1], got {self.success_rate}")

add_instance(upir_id, performance=None)

Add a UPIR instance to this pattern.

Updates instances list and recalculates average performance.

Parameters:

Name Type Description Default
upir_id str

ID of the UPIR instance

 required
performance Dict[str, float]

Performance metrics for this instance

 None
Source code in upir/patterns/pattern.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def add_instance(self, upir_id: str, performance: Dict[str, float] = None):
    """
    Add a UPIR instance to this pattern.

    Updates instances list and recalculates average performance.

    Args:
        upir_id: ID of the UPIR instance
        performance: Performance metrics for this instance
    """
    if upir_id not in self.instances:
        self.instances.append(upir_id)

    # Update average performance if metrics provided
    if performance:
        for metric, value in performance.items():
            if metric not in self.average_performance:
                self.average_performance[metric] = value
            else:
                # Incremental average: new_avg = old_avg + (new_val - old_avg) / count
                count = len(self.instances)
                old_avg = self.average_performance[metric]
                self.average_performance[metric] = old_avg + (value - old_avg) / count

matches(feature_vector, threshold=0.8)

Check if a feature vector matches this pattern.

Uses cosine similarity between the feature vector and the pattern's centroid (if available in template).

Parameters:

Name Type Description Default
feature_vector List[float]

Feature vector to check

 required
threshold float

Similarity threshold (0-1)

 0.8

Returns:

Type Description
bool

True if similarity >= threshold
Source code in upir/patterns/pattern.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def matches(self, feature_vector: List[float], threshold: float = 0.8) -> bool:
    """
    Check if a feature vector matches this pattern.

    Uses cosine similarity between the feature vector and the pattern's
    centroid (if available in template).

    Args:
        feature_vector: Feature vector to check
        threshold: Similarity threshold (0-1)

    Returns:
        True if similarity >= threshold
    """
    if "centroid" not in self.template:
        return False

    import numpy as np
    centroid = np.array(self.template["centroid"])
    vector = np.array(feature_vector)

    # Cosine similarity
    dot_product = np.dot(centroid, vector)
    norm_product = np.linalg.norm(centroid) * np.linalg.norm(vector)

    if norm_product == 0:
        return False

    similarity = dot_product / norm_product
    return similarity >= threshold

to_dict()

Serialize pattern to dictionary.

Returns:

Type Description
Dict[str, Any]

Dictionary representation
Source code in upir/patterns/pattern.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
def to_dict(self) -> Dict[str, Any]:
    """
    Serialize pattern to dictionary.

    Returns:
        Dictionary representation
    """
    return {
        "id": self.id,
        "name": self.name,
        "description": self.description,
        "template": self.template,
        "instances": self.instances,
        "success_rate": self.success_rate,
        "average_performance": self.average_performance,
    }

from_dict(data) classmethod

Deserialize pattern from dictionary.

Parameters:

Name Type Description Default
data Dict[str, Any]

Dictionary containing pattern fields

 required

Returns:

Type Description
Pattern

Pattern instance
Source code in upir/patterns/pattern.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "Pattern":
    """
    Deserialize pattern from dictionary.

    Args:
        data: Dictionary containing pattern fields

    Returns:
        Pattern instance
    """
    return cls(
        id=data["id"],
        name=data["name"],
        description=data["description"],
        template=data.get("template", {}),
        instances=data.get("instances", []),
        success_rate=data.get("success_rate", 0.0),
        average_performance=data.get("average_performance", {}),
    )

__str__()

Human-readable representation.

Source code in upir/patterns/pattern.py 
179
180
181
182
183
184
185
def __str__(self) -> str:
    """Human-readable representation."""
    return (
        f"Pattern({self.name}, "
        f"{len(self.instances)} instances, "
        f"success_rate={self.success_rate:.2f})"
    )

__repr__()

Developer-friendly representation.

Source code in upir/patterns/pattern.py 
187
188
189
190
191
192
193
def __repr__(self) -> str:
    """Developer-friendly representation."""
    return (
        f"Pattern(id='{self.id}', name='{self.name}', "
        f"instances={len(self.instances)}, "
        f"success_rate={self.success_rate:.2f})"
    )

Usage Example

from upir.patterns.pattern import Pattern
from datetime import datetime

# Create pattern
pattern = Pattern(
    id="streaming-etl-001",
    name="Streaming ETL Pattern",
    description="Real-time data pipeline with Pub/Sub -> Beam -> BigQuery",
    template={
        "components": [...],
        "connections": [...],
        "centroid": [...]  # Feature vector
    },
    instances=[],
    success_rate=0.95,
    created_at=datetime.now(),
    updated_at=datetime.now()
)

# Serialize
pattern_json = pattern.to_json()

See Also