Skip to content

aea.helpers.search.models

Useful classes for the OEF search.

Location Objects

class Location()

Data structure to represent locations (i.e. a pair of latitude and longitude).

__init__

def __init__(latitude: float, longitude: float) -> None

Initialize a location.

Arguments:

  • latitude: the latitude of the location.
  • longitude: the longitude of the location.

tuple

@property
def tuple() -> Tuple[float, float]

Get the tuple representation of a location.

distance

def distance(other: "Location") -> float

Get the distance to another location.

Arguments:

  • other: the other location

Returns:

the distance

__eq__

def __eq__(other: Any) -> bool

Compare equality of two locations.

__str__

def __str__() -> str

Get the string representation of the data model.

encode

def encode() -> models_pb2.Query.Location

Encode an instance of this class into a protocol buffer object.

Returns:

the matching protocol buffer object

decode

@classmethod
def decode(cls, location_pb: Any) -> "Location"

Decode a protocol buffer object that corresponds with this class into an instance of this class.

Arguments:

  • location_pb: the protocol buffer object corresponding with this class.

Returns:

A new instance of this class matching the protocol buffer object

AttributeInconsistencyException Objects

class AttributeInconsistencyException(Exception)

Raised when the attributes in a Description are inconsistent.

Inconsistency is defined when values do not meet their respective schema, or if the values are not of an allowed type.

Attribute Objects

class Attribute()

Implements an attribute for an OEF data model.

__init__

def __init__(name: str,
             type_: Type[ATTRIBUTE_TYPES],
             is_required: bool,
             description: str = "") -> None

Initialize an attribute.

Arguments:

  • name: the name of the attribute.
  • type_: the type of the attribute.
  • is_required: whether the attribute is required by the data model.
  • description: an (optional) human-readable description for the attribute.

__eq__

def __eq__(other: Any) -> bool

Compare with another object.

__str__

def __str__() -> str

Get the string representation of the data model.

encode

def encode() -> models_pb2.Query.Attribute

Encode an instance of this class into a protocol buffer object.

Returns:

the matching protocol buffer object

decode

@classmethod
def decode(cls, attribute_pb: models_pb2.Query.Attribute) -> "Attribute"

Decode a protocol buffer object that corresponds with this class into an instance of this class.

Arguments:

  • attribute_pb: the protocol buffer object corresponding with this class.

Returns:

A new instance of this class matching the protocol buffer object

DataModel Objects

class DataModel()

Implements an OEF data model.

__init__

def __init__(name: str,
             attributes: List[Attribute],
             description: str = "") -> None

Initialize a data model.

Arguments:

  • name: the name of the data model.
  • attributes: the attributes of the data model.
  • description: the data model description.

attributes_by_name

@property
def attributes_by_name() -> Dict[str, Attribute]

Get the attributes by name.

__eq__

def __eq__(other: Any) -> bool

Compare with another object.

__str__

def __str__() -> str

Get the string representation of the data model.

encode

def encode() -> models_pb2.Query.DataModel

Encode an instance of this class into a protocol buffer object.

Returns:

the matching protocol buffer object

decode

@classmethod
def decode(cls, data_model_pb: Any) -> "DataModel"

Decode a protocol buffer object that corresponds with this class into an instance of this class.

Arguments:

  • data_model_pb: the protocol buffer object corresponding with this class.

Returns:

A new instance of this class matching the protocol buffer object

generate_data_model

def generate_data_model(
        model_name: str,
        attribute_values: Mapping[str, ATTRIBUTE_TYPES]) -> DataModel

Generate a data model that matches the values stored in this description.

That is, for each attribute (name, value), generate an Attribute. It is assumed that each attribute is required.

Arguments:

  • model_name: the name of the model.
  • attribute_values: the values of each attribute

Returns:

the schema compliant with the values specified.

Description Objects

class Description()

Implements an OEF description.

__init__

def __init__(values: Mapping[str, ATTRIBUTE_TYPES],
             data_model: Optional[DataModel] = None,
             data_model_name: str = "") -> None

Initialize the description object.

Arguments:

  • values: the values in the description.
  • data_model: the data model (optional)
  • data_model_name: the data model name if a datamodel is created on the fly.

values

@property
def values() -> Dict

Get the values.

__eq__

def __eq__(other: Any) -> bool

Compare with another object.

__iter__

def __iter__() -> Iterator

Create an iterator.

__str__

def __str__() -> str

Get the string representation of the description.

encode

@classmethod
def encode(cls, description_pb: Any, description: "Description") -> None

Encode an instance of this class into the protocol buffer object.

The protocol buffer object in the description_protobuf_object argument must be matched with the instance of this class in the 'description_object' argument.

Arguments:

  • description_pb: the protocol buffer object whose type corresponds with this class.
  • description: an instance of this class to be encoded in the protocol buffer object.

decode

@classmethod
def decode(cls, description_pb: Any) -> "Description"

Decode a protocol buffer object that corresponds with this class into an instance of this class.

A new instance of this class must be created that matches the protocol buffer object in the 'description_protobuf_object' argument.

Arguments:

  • description_pb: the protocol buffer object whose type corresponds with this class.

Returns:

A new instance of this class that matches the protocol buffer object in the 'description_protobuf_object' argument.

ConstraintTypes Objects

class ConstraintTypes(Enum)

Types of constraint.

__str__

def __str__() -> str

Get the string representation.

ConstraintType Objects

class ConstraintType()

Type of constraint.

Used with the Constraint class, this class allows to specify constraint over attributes.

Examples:

Equal to three

equal_3 = ConstraintType(ConstraintTypes.EQUAL, 3)

You can also specify a type of constraint by using its string representation, e.g.:

equal_3 = ConstraintType("==", 3) not_equal_london = ConstraintType("!=", "London") less_than_pi = ConstraintType("<", 3.14) within_range = ConstraintType("within", (-10.0, 10.0)) in_a_set = ConstraintType("in", (1, 2, 3)) not_in_a_set = ConstraintType("not_in", ("C", "Java", "Python"))

__init__

def __init__(type_: Union[ConstraintTypes, str], value: Any) -> None

Initialize a constraint type.

Arguments:

  • type_: the type of the constraint. | Either an instance of the ConstraintTypes enum, | or a string representation associated with the type.
  • value: the value that defines the constraint.

Raises:

  • AEAEnforceError: if the type of the constraint is not # noqa: DAR402

check_validity

def check_validity() -> bool

Check the validity of the input provided.

Raises:

  • AEAEnforceError: if the value is not valid wrt the constraint type. # noqa: DAR402

Returns:

boolean to indicate validity

is_valid

def is_valid(attribute: Attribute) -> bool

Check if the constraint type is valid wrt a given attribute.

A constraint type is valid wrt an attribute if the type of its operand(s) is the same of the attribute type.

attribute = Attribute("year", int, True) valid_constraint_type = ConstraintType(ConstraintTypes.GREATER_THAN, 2000) valid_constraint_type.is_valid(attribute) True

valid_constraint_type = ConstraintType(ConstraintTypes.WITHIN, (2000, 2001)) valid_constraint_type.is_valid(attribute) True

The following constraint is invalid: the year is in a string variable, whereas the attribute is defined over integers.

invalid_constraint_type = ConstraintType(ConstraintTypes.GREATER_THAN, "2000") invalid_constraint_type.is_valid(attribute) False

Arguments:

  • attribute: the data model used to check the validity of the constraint type.

Returns:

True if the constraint type is valid wrt the attribute, False otherwise.

get_data_type

def get_data_type() -> Type[ATTRIBUTE_TYPES]

Get the type of the data used to define the constraint type.

For instance:

c = ConstraintType(ConstraintTypes.EQUAL, 1) c.get_data_type()

Returns:

data type

check

def check(value: ATTRIBUTE_TYPES) -> bool

Check if an attribute value satisfies the constraint.

The implementation depends on the constraint type.

Arguments:

  • value: the value to check.

Raises:

  • ValueError: if the constraint type is not recognized.

Returns:

True if the value satisfy the constraint, False otherwise.

__eq__

def __eq__(other: Any) -> bool

Check equality with another object.

__str__

def __str__() -> str

Get the string representation of the constraint type.

encode

def encode() -> Optional[Any]

Encode an instance of this class into a protocol buffer object.

Returns:

the matching protocol buffer object

decode

@classmethod
def decode(cls, constraint_type_pb: Any, category: str) -> "ConstraintType"

Decode a protocol buffer object that corresponds with this class into an instance of this class.

Arguments:

  • constraint_type_pb: the protocol buffer object corresponding with this class.
  • category: the category of the constraint ('relation', 'set', 'range', 'distance).

Returns:

A new instance of this class matching the protocol buffer object

ConstraintExpr Objects

class ConstraintExpr(ABC)

Implementation of the constraint language to query the OEF node.

check

@abstractmethod
def check(description: Description) -> bool

Check if a description satisfies the constraint expression.

Arguments:

  • description: the description to check.

Returns:

True if the description satisfy the constraint expression, False otherwise.

is_valid

@abstractmethod
def is_valid(data_model: DataModel) -> bool

Check whether a constraint expression is valid wrt a data model.

Specifically, check the following conditions: - If all the attributes referenced by the constraints are correctly associated with the Data Model attributes.

Arguments:

  • data_model: the data model used to check the validity of the constraint expression.

Returns:

True if the constraint expression is valid wrt the data model, False otherwise.

check_validity

def check_validity() -> None

Check whether a Constraint Expression satisfies some basic requirements.

Raises:

  • AEAEnforceError: if the object does not satisfy some requirements. # noqa: DAR402

Returns:

None

And Objects

class And(ConstraintExpr)

Implementation of the 'And' constraint expression.

__init__

def __init__(constraints: List[ConstraintExpr]) -> None

Initialize an 'And' expression.

Arguments:

  • constraints: the list of constraints expression (in conjunction).

check

def check(description: Description) -> bool

Check if a value satisfies the 'And' constraint expression.

Arguments:

  • description: the description to check.

Returns:

True if the description satisfy the constraint expression, False otherwise.

is_valid

def is_valid(data_model: DataModel) -> bool

Check whether the constraint expression is valid wrt a data model.

Arguments:

  • data_model: the data model used to check the validity of the constraint expression.

Returns:

True if the constraint expression is valid wrt the data model, False otherwise.

check_validity

def check_validity() -> None

Check whether the Constraint Expression satisfies some basic requirements.

Raises:

  • ValueError: if the object does not satisfy some requirements.

__eq__

def __eq__(other: Any) -> bool

Compare with another object.

encode

def encode() -> models_pb2.Query.ConstraintExpr.And

Encode an instance of this class into a protocol buffer object.

Returns:

the matching protocol buffer object

decode

@classmethod
def decode(cls, and_pb: Any) -> "And"

Decode a protocol buffer object that corresponds with this class into an instance of this class.

Arguments:

  • and_pb: the protocol buffer object corresponding with this class.

Returns:

A new instance of this class matching the protocol buffer object

Or Objects

class Or(ConstraintExpr)

Implementation of the 'Or' constraint expression.

__init__

def __init__(constraints: List[ConstraintExpr]) -> None

Initialize an 'Or' expression.

Arguments:

  • constraints: the list of constraints expressions (in disjunction).

check

def check(description: Description) -> bool

Check if a value satisfies the 'Or' constraint expression.

Arguments:

  • description: the description to check.

Returns:

True if the description satisfy the constraint expression, False otherwise.

is_valid

def is_valid(data_model: DataModel) -> bool

Check whether the constraint expression is valid wrt a data model.

Arguments:

  • data_model: the data model used to check the validity of the constraint expression.

Returns:

True if the constraint expression is valid wrt the data model, False otherwise.

check_validity

def check_validity() -> None

Check whether the Constraint Expression satisfies some basic requirements.

Raises:

  • ValueError: if the object does not satisfy some requirements.

__eq__

def __eq__(other: Any) -> bool

Compare with another object.

encode

def encode() -> models_pb2.Query.ConstraintExpr.Or

Encode an instance of this class into a protocol buffer object.

Returns:

the matching protocol buffer object

decode

@classmethod
def decode(cls, or_pb: Any) -> "Or"

Decode a protocol buffer object that corresponds with this class into an instance of this class.

Arguments:

  • or_pb: the protocol buffer object corresponding with this class.

Returns:

A new instance of this class matching the protocol buffer object

Not Objects

class Not(ConstraintExpr)

Implementation of the 'Not' constraint expression.

__init__

def __init__(constraint: ConstraintExpr) -> None

Initialize a 'Not' expression.

Arguments:

  • constraint: the constraint expression to negate.

check

def check(description: Description) -> bool

Check if a value satisfies the 'Not' constraint expression.

Arguments:

  • description: the description to check.

Returns:

True if the description satisfy the constraint expression, False otherwise.

is_valid

def is_valid(data_model: DataModel) -> bool

Check whether the constraint expression is valid wrt a data model.

Arguments:

  • data_model: the data model used to check the validity of the constraint expression.

Returns:

True if the constraint expression is valid wrt the data model, False otherwise.

__eq__

def __eq__(other: Any) -> bool

Compare with another object.

encode

def encode() -> models_pb2.Query.ConstraintExpr.Not

Encode an instance of this class into a protocol buffer object.

Returns:

the matching protocol buffer object

decode

@classmethod
def decode(cls, not_pb: Any) -> "Not"

Decode a protocol buffer object that corresponds with this class into an instance of this class.

Arguments:

  • not_pb: the protocol buffer object corresponding with this class.

Returns:

A new instance of this class matching the protocol buffer object

Constraint Objects

class Constraint(ConstraintExpr)

The atomic component of a constraint expression.

__init__

def __init__(attribute_name: str, constraint_type: ConstraintType) -> None

Initialize a constraint.

Arguments:

  • attribute_name: the name of the attribute to be constrained.
  • constraint_type: the constraint type.

check

def check(description: Description) -> bool

Check if a description satisfies the constraint. The implementation depends on the type of the constraint.

Arguments:

  • description: the description to check.

Returns:

True if the description satisfies the constraint, False otherwise. Examples: >>> attr_author = Attribute("author" , str, True, "The author of the book.") >>> attr_year = Attribute("year", int, True, "The year of publication of the book.") >>> attr_genre = Attribute("genre", str, True, "The genre of the book.") >>> c1 = Constraint("author", ConstraintType("==", "Stephen King")) >>> c2 = Constraint("year", ConstraintType(">", 1990)) >>> c3 = Constraint("genre", ConstraintType("in", ("horror", "science_fiction"))) >>> book_1 = Description({"author": "Stephen King", "year": 1991, "genre": "horror"}) >>> book_2 = Description({"author": "George Orwell", "year": 1948, "genre": "horror"})

The "author" attribute instantiation satisfies the constraint, so the result is True.

>>> c1.check(book_1)
True

Here, the "author" does not satisfy the constraints. Hence, the result is False.

>>> c1.check(book_2)
False

In this case, there is a missing field specified by the query, that is "year"
So the result is False, even in the case it is not required by the schema:

>>> c2.check(Description({"author": "Stephen King"}))
False

If the type of some attribute of the description is not correct, the result is False.
In this case, the field "year" has a string instead of an integer:

>>> c2.check(Description({"author": "Stephen King", "year": "1991"}))
False

>>> c3.check(Description({"author": "Stephen King", "genre": False}))
False

is_valid

def is_valid(data_model: DataModel) -> bool

Check whether the constraint expression is valid wrt a data model.

Arguments:

  • data_model: the data model used to check the validity of the constraint expression.

Returns:

True if the constraint expression is valid wrt the data model, False otherwise.

__eq__

def __eq__(other: Any) -> bool

Compare with another object.

__str__

def __str__() -> str

Get the string representation of the constraint.

encode

def encode() -> models_pb2.Query.ConstraintExpr.Constraint

Encode an instance of this class into a protocol buffer object.

Returns:

the matching protocol buffer object

decode

@classmethod
def decode(cls, constraint_pb: Any) -> "Constraint"

Decode a protocol buffer object that corresponds with this class into an instance of this class.

Arguments:

  • constraint_pb: the protocol buffer object corresponding with this class.

Returns:

A new instance of this class matching the protocol buffer object

Query Objects

class Query()

This class lets you build a query for the OEF.

__init__

def __init__(constraints: List[ConstraintExpr],
             model: Optional[DataModel] = None) -> None

Initialize a query.

Arguments:

  • constraints: a list of constraint expressions.
  • model: the data model that the query refers to.

check

def check(description: Description) -> bool

Check if a description satisfies the constraints of the query.

The constraints are interpreted as conjunction.

Arguments:

  • description: the description to check.

Returns:

True if the description satisfies all the constraints, False otherwise.

is_valid

def is_valid(data_model: Optional[DataModel]) -> bool

Given a data model, check whether the query is valid for that data model.

Arguments:

  • data_model: optional datamodel

Returns:

True if the query is compliant with the data model, False otherwise.

check_validity

def check_validity() -> None

Check whether the` object is valid.

Raises:

  • ValueError: if the query does not satisfy some sanity requirements.

__eq__

def __eq__(other: Any) -> bool

Compare with another object.

__str__

def __str__() -> str

Get the string representation of the constraint.

encode

@classmethod
def encode(cls, query_pb: Any, query: "Query") -> None

Encode an instance of this class into the protocol buffer object.

The protocol buffer object in the query_protobuf_object argument must be matched with the instance of this class in the 'query_object' argument.

Arguments:

  • query_pb: the protocol buffer object wrapping an object that corresponds with this class.
  • query: an instance of this class to be encoded in the protocol buffer object.

decode

@classmethod
def decode(cls, query_pb: Any) -> "Query"

Decode a protocol buffer object that corresponds with this class into an instance of this class.

A new instance of this class must be created that matches the protocol buffer object in the 'query_protobuf_object' argument.

Arguments:

  • query_pb: the protocol buffer object whose type corresponds with this class.

Returns:

A new instance of this class that matches the protocol buffer object in the 'query_protobuf_object' argument.

haversine

def haversine(lat1: float, lon1: float, lat2: float, lon2: float) -> float

Compute the Haversine distance between two locations (i.e. two pairs of latitude and longitude).

Arguments:

  • lat1: the latitude of the first location.
  • lon1: the longitude of the first location.
  • lat2: the latitude of the second location.
  • lon2: the longitude of the second location.

Returns:

the Haversine distance.