Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def from_dictionary(cls, arguments, entityset, dependencies, primitives_deserializer):
base_features = [dependencies[name] for name in arguments['base_features']]
relationship_path = [Relationship.from_dictionary(r, entityset)
for r in arguments['relationship_path']]
parent_entity = relationship_path[0].parent_entity
relationship_path = RelationshipPath([(False, r) for r in relationship_path])
primitive = primitives_deserializer.deserialize_primitive(arguments['primitive'])
use_previous_data = arguments['use_previous']
use_previous = use_previous_data and Timedelta.from_dictionary(use_previous_data)
where_name = arguments['where']
where = where_name and dependencies[where_name]
return cls(base_features=base_features, parent_entity=parent_entity, primitive=primitive, relationship_path=relationship_path,
use_previous=use_previous, where=where, name=arguments['name'])
def get_backward_entities(self, entity_id, deep=False):
"""
Get entities that are in a backward relationship with entity
Args:
entity_id (str): Id entity of entity to search from.
deep (bool): if True, recursively find backward entities.
Yields a tuple of (descendent_id, path from entity_id to descendant).
"""
for relationship in self.get_backward_relationships(entity_id):
child_eid = relationship.child_entity.id
direct_path = RelationshipPath([(False, relationship)])
yield child_eid, direct_path
if deep:
sub_entities = self.get_backward_entities(child_eid, deep=True)
for sub_eid, path in sub_entities:
yield sub_eid, direct_path + path
def get_forward_entities(self, entity_id, deep=False):
"""
Get entities that are in a forward relationship with entity
Args:
entity_id (str): Id entity of entity to search from.
deep (bool): if True, recursively find forward entities.
Yields a tuple of (descendent_id, path from entity_id to descendant).
"""
for relationship in self.get_forward_relationships(entity_id):
parent_eid = relationship.parent_entity.id
direct_path = RelationshipPath([(True, relationship)])
yield parent_eid, direct_path
if deep:
sub_entities = self.get_forward_entities(parent_eid, deep=True)
for sub_eid, path in sub_entities:
yield sub_eid, direct_path + path
def _build_feature_trie(self):
"""
Build the feature trie by adding the target features and their dependencies recursively.
"""
feature_trie = Trie(default=lambda: (False, set(), set()),
path_constructor=RelationshipPath)
for f in self.target_features:
self._add_feature_to_trie(feature_trie,
f,
self.approximate_feature_trie)
return feature_trie
def __eq__(self, other):
return isinstance(other, RelationshipPath) and \
self._relationships_with_direction == other._relationships_with_direction
def __init__(self, variable, name=None):
entity_id = variable.entity_id
self.variable = variable.entityset.metadata[entity_id][variable.id]
self.return_type = type(variable)
super(IdentityFeature, self).__init__(entity=variable.entity,
base_features=[],
relationship_path=RelationshipPath([]),
primitive=PrimitiveBase,
name=name)
.has_unique_forward_path(self.child_entity.id, parent_entity.id)
else:
paths = parent_entity.entityset \
.find_backward_paths(parent_entity.id, self.child_entity.id)
first_path = next(paths, None)
if not first_path:
raise RuntimeError('No backward path from "%s" to "%s" found.'
% (parent_entity.id, self.child_entity.id))
# Check for another path.
elif next(paths, None):
message = "There are multiple possible paths to the base entity. " \
"You must specify a relationship path."
raise RuntimeError(message)
relationship_path = RelationshipPath([(False, r) for r in first_path])
path_is_unique = True
return relationship_path, path_is_unique