from collections import Counter
from dataclasses import dataclass, field
from enum import Enum
from pathlib import Path
from typing import Dict, Union
from warnings import warn
from simprov import Entity, Activity, Agent
from simprov.exceptions import InvalidEntitySpecificationException, EntitySpecificationNotFoundException, \
PrimaryKeyAttributeNotDefinedException, ActivitySpecificationNotFoundException, ActivityAlreadyDefinedException, \
EntityAlreadyDefinedException, InvalidActivityException, InvalidActivitySpecificationException, \
InvalidSpecificationException, AgentSpecificationNotFoundException
def _read_yaml_file(file_path):
import yaml
with open(file_path, "r") as file_handler:
yaml_content = yaml.safe_load(file_handler)
return yaml_content
[docs]
@dataclass
class EntitySpecification:
"""Represents a specification of an entity.
:ivar str name:
The name of the entity for whic°h the specification should be used, e.g., simulation model or simulation experiement.
:ivar List[str] required_attributes:
A list of attributes names that are required for an entiy
:ivar List[str] primary_key_attributes:
A list of attributes that form the primary key of an entity.
:ivar List[str] attributes:
A list of all attribute names of an entity.
It also contains the required attributes as well as the primary key attributes.
:ivar Dict style_info:
Holds the style information extracted from the specification file.
It can be used to store information about how an entity should be rendered in the webview.
"""
name: str
required_attributes: list = field(default_factory=list)
attributes: list = field(default_factory=list)
primary_key_attributes: list = field(default_factory=list)
style_info: dict = field(default_factory=dict)
@dataclass
class AgentSpecification:
"""Represents a specification of an agent..
:ivar str name:
The name of the agent for which the specification should be used, e.g., simulator or python environment.
:ivar List[str] required_attributes:
A list of attributes names that are required for an agent
:ivar List[str] primary_key_attributes:
A list of attributes that form the primary key of an agent.
:ivar List[str] attributes:
A list of all attribute names of an agent.
It also contains the required attributes as well as the primary key attributes.
:ivar Dict style_info:
Holds the style information extracted from the specification file.
It can be used to store information about how an agent should be rendered in the webview.
"""
name: str
required_attributes: list = field(default_factory=list)
attributes: list = field(default_factory=list)
primary_key_attributes: list = field(default_factory=list)
style_info: dict = field(default_factory=dict)
[docs]
class OccurenceModifier(Enum):
"""Represents a modifier that signals how often an entity of a name can be used or generated by an activity.
Possible Values:
- ``SINGLE`` - An entity has occur exactly once
- ``ZERO_OR_ONE`` - An entity is optional
- ``ONE_OR_MORE`` - An entity can occur arbitrary often
- ``ZERO_OR_MORE`` - An entity has to occur at least once
"""
SINGLE = 0,
ZERO_OR_ONE = 1,
ONE_OR_MORE = 2,
ZERO_OR_MORE = 3,
[docs]
@dataclass
class ActivitySpecification:
"""Represents a specification of an activity.
:ivar str name:
The name of the activity for which the specification should be used.
:ivar Dict[str,OccurenceModifier] used_entities:
A mapping from the entity names of entities that were used by the activity to a modifier.
:ivar Dict[str, OccurenceModifier] generated_entities:
A mapping from the entity names of entities that were generated by the activity to a modifier.
"""
name: str
used_entities: Dict[str, OccurenceModifier] = field(default_factory=dict)
generated_entities: Dict[str, OccurenceModifier] = field(default_factory=dict)
associated_agents: Dict[str, OccurenceModifier] = field(default_factory=dict)
[docs]
class SpecificationManager:
""" The specification manager loads and stores all entity and activity specifications.
:param Union[str, Path], optional specification_path:
When provided the specifications are loaded from the file.
:ivar Dict[str, EntitySpecification] entity_specifications:
A mapping from entity names to their corresponding specifications.
:ivar Dict[str, ActivitySpecification] activity_specifications:
A mapping from activity names to their corresponding specifications.
"""
def __init__(self, specification_path: Union[str, Path] = None):
super().__init__()
self.entity_specifications: Dict[str, EntitySpecification] = {}
self.activity_specifications: Dict[str, ActivitySpecification] = {}
self.agent_specifications: Dict[str, AgentSpecification] = {}
if specification_path:
self.load_specification_file(specification_path)
[docs]
def get_entity_specification(self, entity_name: str) -> EntitySpecification:
""" Gets the entity specification for an entity name.
:param str entity_name:
The entity name.
:rtype: EntitySpecification
:return: The entity specification
:raises EntitySpecificationNotFoundException:
If a specification can not be found.
"""
entity_spec = self.entity_specifications.get(entity_name, None)
if entity_spec is None:
raise EntitySpecificationNotFoundException(f"Can't find specification for entity \"{entity_name}\"")
return entity_spec
[docs]
def get_activity_specification(self, activity_name: str) -> ActivitySpecification:
"""Returns the corresponding activity specification for an activity name.
:param str activity_name:
The activity name.
:rtype: ActivitySpecification
:return: The activity specification
:raises ActivitySpecificationNotFoundException:
If a specification can not be found.
"""
activity_specification = self.activity_specifications.get(activity_name, None)
if activity_specification is None:
raise ActivitySpecificationNotFoundException(f"Can't find specification for activity \"{activity_name}\"")
return activity_specification
[docs]
def normalize_activity(self, activity: Activity):
"""Normalizes an activity.
Iterates over all used and generated entities of activity and normalizes them.
"""
for entity in activity.entities:
self.normalize_entity(entity)
for agent in activity.associated_agents:
self.normalize_agent(agent)
return activity
[docs]
def normalize_entity(self, entity: Entity):
"""Normalizes an entity.
Gets the entity specification to set all missing entity attributes to ``None``, to set the primary key
and to inject the meta information.
:param Entity entity:
The entity.
"""
entity_spec: EntitySpecification = self.get_entity_specification(entity.name)
for attribute in entity_spec.attributes:
if attribute not in entity.attributes:
entity.attributes[attribute] = None
entity.primary_key = tuple([entity.attributes[attribute] for attribute in entity_spec.primary_key_attributes])
entity.meta_information = entity_spec.style_info
[docs]
def normalize_agent(self, agent: Agent):
"""Normalizes an agent.
Gets the agent specification to set all missing entity attributes to ``None``, to set the primary key
and to inject the meta information.
:param Agent agent:
The agent.
"""
agent_spec: AgentSpecification = self.get_agent_specification(agent.name)
for attribute in agent_spec.attributes:
if attribute not in agent.attributes:
agent.attributes[attribute] = None
agent.primary_key = tuple([agent.attributes[attribute] for attribute in agent_spec.primary_key_attributes])
agent.meta_information = agent_spec.style_info
[docs]
def validate_entity(self, entity: Entity):
"""Validates whether an entity correspond to its entity specification.
:param Entity entity:
The entity.
:raises PrimaryKeyAttributeNotFound:
If a primary key attribute is not in included in the entity attributes
:raises PrimaryKeyAttributeNotDefined:
If the value of a primary key attribute is `None`
"""
entity_spec = self.get_entity_specification(entity.name)
for primary_key_attribute in entity_spec.primary_key_attributes:
if primary_key_attribute not in entity.attributes:
raise PrimaryKeyAttributeNotDefinedException(
f"Can't find primary key attribute \"{primary_key_attribute}\" in entity {entity.name}")
if entity.attributes[primary_key_attribute] is None:
raise PrimaryKeyAttributeNotDefinedException(
f"Primary Key Attributes \"{primary_key_attribute}\" for Entity \"{entity.name}\" is None. Check rules!")
[docs]
def validate_agent(self, agent: Agent):
"""Validates whether an agent correspond to its agent specification.
:param Agent agent:
The agent.
:raises PrimaryKeyAttributeNotFound:
If a primary key attribute is not in included in the entity attributes
:raises PrimaryKeyAttributeNotDefined:
If the value of a primary key attribute is `None`
"""
agent_spec = self.get_agent_specification(agent.name)
for primary_key_attribute in agent_spec.primary_key_attributes:
if primary_key_attribute not in agent.attributes:
raise PrimaryKeyAttributeNotDefinedException(
f"Can't find primary key attribute \"{primary_key_attribute}\" in agent {agent.name}")
if agent.attributes[primary_key_attribute] is None:
raise PrimaryKeyAttributeNotDefinedException(
f"Primary Key Attributes \"{primary_key_attribute}\" for agent \"{agent.name}\" is None. Check rules!")
[docs]
def validate_activity(self, activity: Activity):
""" Validates whether the activity corresponds to its activity specification.
:param Activity activity:
The activity.
:raises InvalidActivityException:
If the names or count of the entity names of the activity does not match the activity specification.
"""
activity_specification = self.get_activity_specification(activity.name)
activity_usage_entities = [entity.name for entity in activity.used_entities]
activity_generated_entities = [entity.name for entity in activity.generated_entities]
activity_associated_agents = [agent.name for agent in activity.associated_agents]
counted_usage_entity_names = Counter(activity_usage_entities)
counted_generated_entity_names = Counter(activity_generated_entities)
counted_associated_agent_names = Counter(activity_associated_agents)
for counted_usage_entity_name in counted_usage_entity_names:
if counted_usage_entity_name not in activity_specification.used_entities:
raise InvalidActivityException(
f"Activity \"{activity.name}\" uses an entity \"{counted_usage_entity_name}\" which is not part of the specification.")
for counted_generated_entity_name in counted_generated_entity_names:
if counted_generated_entity_name not in activity_specification.generated_entities:
raise InvalidActivityException(
f"Activity \"{activity.name}\" generates an entity \"{counted_generated_entity_name}\" which is not part of the specification.")
for (used_entity_name, used_entity_modifier) in activity_specification.used_entities.items():
count = counted_usage_entity_names[used_entity_name]
if used_entity_modifier == OccurenceModifier.SINGLE:
if count != 1:
raise InvalidActivityException(
f"Activity Specification \"{activity_specification.name}\" requires exactly one used entity of type \"{used_entity_name}\"")
elif used_entity_modifier == OccurenceModifier.ZERO_OR_ONE:
if count > 1:
raise InvalidActivityException(
f"Activity Specification \"{activity_specification.name}\" requires zero or one used entity of type \"{used_entity_name}\"")
elif used_entity_modifier == OccurenceModifier.ONE_OR_MORE:
if count < 1:
raise InvalidActivityException(
f"Activity Specification \"{activity_specification.name}\" requires one or more used entities of type \"{used_entity_name}\"")
for (generated_entity_name, generated_entity_modifier) in activity_specification.generated_entities.items():
count = counted_generated_entity_names[generated_entity_name]
if generated_entity_modifier == OccurenceModifier.SINGLE:
if count != 1:
raise InvalidActivityException(
f"Activity Specification \"{activity_specification.name}\" requires exactly one generated entity of type \"{generated_entity_name}\"")
elif generated_entity_modifier == OccurenceModifier.ZERO_OR_ONE:
if count > 1:
raise InvalidActivityException(
f"Activity Specification \"{activity_specification.name}\" requires zero or one generated entity of type \"{generated_entity_name}\"")
elif generated_entity_modifier == OccurenceModifier.ONE_OR_MORE:
if count < 1:
raise InvalidActivityException(
f"Activity Specification \"{activity_specification.name}\" requires one or more generated entities of type \"{generated_entity_name}\"")
for (associated_agent_name, associated_agent_modifier) in activity_specification.associated_agents.items():
count = counted_associated_agent_names[associated_agent_name]
if associated_agent_modifier == OccurenceModifier.SINGLE:
if count != 1:
raise InvalidActivityException(
f"Activity Specification \"{activity_specification.name}\" requires exactly one generated agent of type \"{associated_agent_name}\"")
elif associated_agent_modifier == OccurenceModifier.ZERO_OR_ONE:
if count > 1:
raise InvalidActivityException(
f"Activity Specification \"{activity_specification.name}\" requires zero or one generated agent of type \"{associated_agent_name}\"")
elif associated_agent_modifier == OccurenceModifier.ONE_OR_MORE:
if count < 1:
raise InvalidActivityException(
f"Activity Specification \"{activity_specification.name}\" requires one or more generated agent of type \"{associated_agent_name}\"")
for entity in activity.entities:
self.validate_entity(entity)
for agent in activity.associated_agents:
self.validate_agent(agent)
def _process_entity_specification(self, specification):
entity_specification = self._build_entitiy_specification(specification)
self.entity_specifications[entity_specification.name] = entity_specification
def _process_agent_specification(self, specification):
agent_specification = self._build_agent_specification(specification)
self.agent_specifications[agent_specification.name] = agent_specification
def _process_activity_specification(self, specification):
provenance_pattern = self._build_activity_specification(specification)
self.activity_specifications[provenance_pattern.name] = provenance_pattern
[docs]
def load_specification_file(self, file_path: Union[str, Path]):
""" Loads the specifications from a given file.
:param Union[str, Path] file_path:
The file path.
:raises InvalidSpecificationException:
If an entry in the specification file is neither an entity nor an activity specification.
"""
assert(Path(file_path).exists())
yaml_content = _read_yaml_file(file_path)
backlog = []
# Parse Activity Specifications
for thing in yaml_content.items():
_, sub_mappings = thing
if "generation" in sub_mappings and "usage" in sub_mappings:
self._process_activity_specification(thing)
else:
backlog.append(thing)
# if "attributes" in sub_mappings:
# self._process_entity_specification(thing)
# elif "generation" in sub_mappings and "usage" in sub_mappings:
# self._process_activity_specification(thing)
# else:
# raise InvalidSpecificationException(
# f"Can't build neither an entity nor activity specification from \"{thing}\". Please check your syntax in your specification file.")
# Try to derive the type of the remaining items
valid_entity_names, valid_agent_names = self._extract_valid_entity_and_agent_names()
seen_entity_names = set()
seen_agent_names = set()
if len(valid_agent_names & valid_entity_names) != 0:
raise InvalidSpecificationException(
f"Ambiguous names for entities and agent: \"{valid_agent_names & valid_entity_names}\"")
for item in backlog:
name, sub_mappings = item
if "attributes" not in sub_mappings:
raise InvalidSpecificationException(
f"Can't build neither an entity nor activity specification from \"{thing}\". Please check syntax in your specification file.")
if name in valid_entity_names:
self._process_entity_specification(item)
seen_entity_names.add(name)
elif name in valid_agent_names:
self._process_agent_specification(item)
seen_agent_names.add(name)
else:
warn(
f"Cannot determine specification type for \"{name}\". Specification will be assumed to be an entity specification.")
self._process_entity_specification(item)
if len(valid_entity_names - seen_entity_names) != 0:
raise InvalidSpecificationException(
f"There are entities declared as used or generated by an activity but no specifications are found: {valid_entity_names - seen_entity_names}")
if len(valid_agent_names - seen_agent_names) != 0:
raise InvalidSpecificationException(
f"There are agents declared as associated with an activity but no specifications are found: {valid_agent_names - seen_agent_names}")
def _parse_entity_name(self, entity_name):
modifier = OccurenceModifier.SINGLE
has_modifier = lambda name: name[-1] in ["?", "*", "+"]
is_question_mark_modifier = lambda name: entity_name[-1] == "?"
is_star_modifier = lambda name: name[-1] == "*"
is_plus_modifier = lambda name: name[-1] == "+"
if has_modifier(entity_name):
if is_question_mark_modifier(entity_name):
modifier = OccurenceModifier.ZERO_OR_ONE
elif is_star_modifier(entity_name):
modifier = OccurenceModifier.ZERO_OR_MORE
elif is_plus_modifier(entity_name):
modifier = OccurenceModifier.ONE_OR_MORE
entity_name = entity_name[:-1]
return entity_name, modifier
def _build_activity_specification(self, raw_activity_specification) -> ActivitySpecification:
name = raw_activity_specification[0]
if name in self.activity_specifications:
raise ActivityAlreadyDefinedException(
f"Activity \"{name}\" is already defined. The definition might be more than once in your specifications.")
used_and_generated_entities = raw_activity_specification[1]
used_entities = {}
generated_entities = {}
associated_agents = {}
if "generation" not in used_and_generated_entities:
raise InvalidActivitySpecificationException(
f"The activity specification for \"{name}\" lacks \"generation\" specification.")
if "usage" not in used_and_generated_entities:
raise InvalidActivitySpecificationException(
f"The activity specification for \"{name}\" lacks \"usage\" specification.")
generated_entities_names = used_and_generated_entities.get("generation", [])
used_entities_names = used_and_generated_entities.get("usage", [])
associated_agent_names = used_and_generated_entities.get("association", [])
for entity in used_entities_names:
(entity_name, entity_modifier) = self._parse_entity_name(entity)
used_entities[entity_name] = entity_modifier
for entity in generated_entities_names:
(entity_name, entity_modifier) = self._parse_entity_name(entity)
generated_entities[entity_name] = entity_modifier
for entity in associated_agent_names:
(agent_name, agent_modifier) = self._parse_entity_name(entity)
associated_agents[agent_name] = agent_modifier
activity_specification = ActivitySpecification(name, used_entities, generated_entities, associated_agents)
return activity_specification
def _build_agent_specification(self, raw_agent_specification) -> AgentSpecification:
name, entity_mapping = raw_agent_specification
if name in self.agent_specifications:
raise EntityAlreadyDefinedException(
f"Agent \"{name}\" is already defined. The definition might be more than once in your specifications.")
has_modifier = lambda attribute: attribute[-1] in ["!", "$"]
is_required_attribute = lambda attribute: attribute[-1] == "!"
is_primary_key_attribute = lambda attribute: attribute[-1] == "$"
attributes = []
required_attributes = []
primary_key_attributes = []
for attribute in entity_mapping["attributes"]:
if has_modifier(attribute):
attribute_name = attribute[:-1]
if is_primary_key_attribute(attribute):
primary_key_attributes.append(attribute_name)
required_attributes.append(attribute_name)
elif is_required_attribute(attribute):
required_attributes.append(attribute_name)
attributes.append(attribute_name)
else:
attributes.append(attribute)
if len(primary_key_attributes) == 0:
raise InvalidEntitySpecificationException(
f"Entity specification {raw_agent_specification} misses primary key.")
agent_specification = AgentSpecification(name, required_attributes, attributes,
primary_key_attributes)
agent_specification.style_info = entity_mapping.get("style", {})
return agent_specification
def _build_entitiy_specification(self, raw_entity_specification) -> EntitySpecification:
name, entity_mapping = raw_entity_specification
if name in self.entity_specifications:
raise EntityAlreadyDefinedException(
f"Entity \"{name}\" is already defined. The definition might be more than once in your specifications.")
has_modifier = lambda attribute: attribute[-1] in ["!", "$"]
is_required_attribute = lambda attribute: attribute[-1] == "!"
is_primary_key_attribute = lambda attribute: attribute[-1] == "$"
attributes = []
required_attributes = []
primary_key_attributes = []
for attribute in entity_mapping["attributes"]:
if has_modifier(attribute):
attribute_name = attribute[:-1]
if is_primary_key_attribute(attribute):
primary_key_attributes.append(attribute_name)
required_attributes.append(attribute_name)
elif is_required_attribute(attribute):
required_attributes.append(attribute_name)
attributes.append(attribute_name)
else:
attributes.append(attribute)
if len(primary_key_attributes) == 0:
raise InvalidEntitySpecificationException(
f"Entity specification {raw_entity_specification} misses primary key.")
entity_specification = EntitySpecification(name, required_attributes, attributes,
primary_key_attributes)
entity_specification.style_info = entity_mapping.get("style", {})
return entity_specification
[docs]
def is_editable(self, entity_name: str, attribute_name: str) -> bool:
"""Checks whether an attribute of an entity is editable by the user.
All attributes are editable unless they are part of the primary key attributes.
:param str entity_name:
The entity name.
:param str attribute_name:
The attribute name.
:rtype bool:
:return:
`True` if the attribute is editable, `False` otherwise
"""
spec = self.get_entity_specification(entity_name)
return not attribute_name in spec.primary_key_attributes
def _extract_valid_entity_and_agent_names(self):
valid_agent_names = set()
valid_entity_names = set()
for spec in self.activity_specifications.values():
valid_agent_names |= spec.associated_agents.keys()
valid_entity_names |= spec.generated_entities.keys() | spec.used_entities
return valid_entity_names, valid_agent_names
[docs]
def get_agent_specification(self, agent_name: str) -> AgentSpecification:
""" Gets the agent specification for an agent name.
:param str agent_name:
The agent name.
:rtype: AgentSpecification
:return: The agent specification
:raises AgentSpecificationNotFoundException:
If a specification can not be found.
"""
agent_spec = self.agent_specifications.get(agent_name, None)
if agent_spec is None:
raise AgentSpecificationNotFoundException(f"Can't find specification for entity \"{agent_name}\"")
return agent_spec