teaching/muMLE
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

cleanup the od api interface

Browse source
This commit is contained in:
robbe 2025-06-27 12:15:19 +02:00
parent 04a17f6ac8
commit 9eea3618d0
6 changed files with 69 additions and 47 deletions
Download patch file Download diff file Expand all files Collapse all files

81
api/od.py
View file

@ -6,8 +6,7 @@ from services.primitives.integer_type import Integer
from services.primitives.string_type import String
from services.primitives.actioncode_type import ActionCode
from uuid import UUID
from typing import Optional
from util.timer import Timer
from typing import Optional, Any
NEXT_ID = 0
@ -42,10 +41,10 @@ class ODAPI:
self.create_string_value = self.od.create_string_value
self.create_actioncode_value = self.od.create_actioncode_value
self.__recompute_mappings()
self.recompute_mappings()
# Called after every change - makes querying faster but modifying slower
def __recompute_mappings(self):
def recompute_mappings(self):
self.m_obj_to_name = build_name_mapping(self.state, self.m)
self.mm_obj_to_name = build_name_mapping(self.state, self.mm)
self.type_to_objs = { type_name : set() for type_name in self.bottom.read_keys(self.mm)}
@ -60,25 +59,33 @@ class ODAPI:
def get_value(self, obj: UUID):
return od.read_primitive_value(self.bottom, obj, self.mm)[0]
def get_target(self, link: UUID):
def get_target(self, link: UUID) -> UUID:
return self.bottom.read_edge_target(link)
def get_source(self, link: UUID):
def get_source(self, link: UUID) -> UUID:
return self.bottom.read_edge_source(link)
def get_slot(self, obj: UUID, attr_name: str):
def get_slot(self, obj: UUID, attr_name: str) -> UUID:
slot = self.od.get_slot(obj, attr_name)
if slot == None:
raise NoSuchSlotException(f"Object '{self.m_obj_to_name[obj]}' has no slot '{attr_name}'")
return slot
def get_slot_link(self, obj: UUID, attr_name: str):
def get_slot_link(self, obj: UUID, attr_name: str) -> UUID:
return self.od.get_slot_link(obj, attr_name)
# Parameter 'include_subtypes': whether to include subtypes of the given association
def get_outgoing(self, obj: UUID, assoc_name: str, include_subtypes=True):
def get_outgoing(self, obj: UUID, assoc_name: str, include_subtypes=True) -> list[UUID]:
outgoing = self.bottom.read_outgoing_edges(obj)
result = []
return self.filter_edges_by_type(outgoing, assoc_name, include_subtypes)
# Parameter 'include_subtypes': whether to include subtypes of the given association
def get_incoming(self, obj: UUID, assoc_name: str, include_subtypes=True):
incoming = self.bottom.read_incoming_edges(obj)
return self.filter_edges_by_type(incoming, assoc_name, include_subtypes)
def filter_edges_by_type(self, outgoing: list[UUID], assoc_name: str, include_subtypes=True) -> list[UUID]:
result: list[UUID] = []
for o in outgoing:
try:
type_of_outgoing_link = self.get_type_name(o)
@ -89,23 +96,8 @@ class ODAPI:
result.append(o)
return result
# Parameter 'include_subtypes': whether to include subtypes of the given association
def get_incoming(self, obj: UUID, assoc_name: str, include_subtypes=True):
incoming = self.bottom.read_incoming_edges(obj)
result = []
for i in incoming:
try:
type_of_incoming_link = self.get_type_name(i)
except:
continue # OK, not all edges are typed
if (include_subtypes and self.cdapi.is_subtype(super_type_name=assoc_name, sub_type_name=type_of_incoming_link)
or not include_subtypes and type_of_incoming_link == assoc_name):
result.append(i)
return result
# Returns list of tuples (name, obj)
def get_all_instances(self, type_name: str, include_subtypes=True):
def get_all_instances(self, type_name: str, include_subtypes=True) -> list[UUID]:
if include_subtypes:
all_types = self.cdapi.transitive_sub_types[type_name]
else:
@ -127,7 +119,7 @@ class ODAPI:
else:
raise Exception(f"Couldn't find name of {obj} - are you sure it exists in the (meta-)model?")
def get(self, name: str):
def get(self, name: str) -> UUID:
results = self.bottom.read_outgoing_elements(self.m, name)
if len(results) == 1:
return results[0]
@ -136,10 +128,10 @@ class ODAPI:
else:
raise Exception(f"No such element in model: '{name}'")
def get_type_name(self, obj: UUID):
def get_type_name(self, obj: UUID) -> str:
return self.get_name(self.get_type(obj))
def is_instance(self, obj: UUID, type_name: str, include_subtypes=True):
def is_instance(self, obj: UUID, type_name: str, include_subtypes=True) -> bool:
typ = self.cdapi.get_type(type_name)
types = set(typ) if not include_subtypes else self.cdapi.transitive_sub_types[type_name]
for type_of_obj in self.bottom.read_outgoing_elements(obj, "Morphism"):
@ -147,18 +139,21 @@ class ODAPI:
return True
return False
def delete(self, obj: UUID):
def delete(self, obj: UUID) -> None:
self.bottom.delete_element(obj)
self.__recompute_mappings()
self.recompute_mappings()
# Does the the object have the given attribute?
def has_slot(self, obj: UUID, attr_name: str):
return self.od.get_slot_link(obj, attr_name) != None
def has_slot(self, obj: UUID, attr_name: str) -> bool:
class_name = self.get_name(self.get_type(obj))
if self.od.get_attr_link_name(class_name, attr_name) is None:
return False
return self.od.get_slot_link(obj, attr_name) is not None
def get_slots(self, obj: UUID) -> list[str]:
return [attr_name for attr_name, _ in self.od.get_slots(obj)]
def get_slot_value(self, obj: UUID, attr_name: str):
def get_slot_value(self, obj: UUID, attr_name: str) -> Any:
slot = self.get_slot(obj, attr_name)
return self.get_value(slot)
@ -171,14 +166,14 @@ class ODAPI:
# Returns the given default value if the slot does not exist on the object.
# The attribute must exist in the object's class, or an exception will be thrown.
# The slot may not exist however, if the attribute is defined as 'optional' in the class.
def get_slot_value_default(self, obj: UUID, attr_name: str, default: any):
def get_slot_value_default(self, obj: UUID, attr_name: str, default: any) -> any:
try:
return self.get_slot_value(obj, attr_name)
except NoSuchSlotException:
return default
# create or update slot value
def set_slot_value(self, obj: UUID, attr_name: str, new_value: any, is_code=False):
def set_slot_value(self, obj: UUID, attr_name: str, new_value: any, is_code=False) -> None:
obj_name = self.get_name(obj)
link_name = f"{obj_name}_{attr_name}"
@ -193,7 +188,7 @@ class ODAPI:
new_target = self.create_primitive_value(target_name, new_value, is_code)
slot_type = self.cdapi.find_attribute_type(self.get_type_name(obj), attr_name)
new_link = self.od._create_link(link_name, slot_type, obj, new_target)
self.__recompute_mappings()
self.recompute_mappings()
def create_primitive_value(self, name: str, value: any, is_code=False):
# watch out: in Python, 'bool' is subtype of 'int'
@ -209,7 +204,7 @@ class ODAPI:
tgt = self.create_string_value(name, value)
else:
raise Exception("Unimplemented type "+value)
self.__recompute_mappings()
self.recompute_mappings()
return tgt
def overwrite_primitive_value(self, name: str, value: any, is_code=False):
@ -228,7 +223,7 @@ class ODAPI:
else:
raise Exception("Unimplemented type "+value)
def create_link(self, link_name: Optional[str], assoc_name: str, src: UUID, tgt: UUID):
def create_link(self, link_name: Optional[str], assoc_name: str, src: UUID, tgt: UUID) -> UUID:
global NEXT_ID
types = self.bottom.read_outgoing_elements(self.mm, assoc_name)
if len(types) == 0:
@ -240,12 +235,12 @@ class ODAPI:
link_name = f"__{assoc_name}{NEXT_ID}"
NEXT_ID += 1
link_id = self.od._create_link(link_name, typ, src, tgt)
self.__recompute_mappings()
self.recompute_mappings()
return link_id
def create_object(self, object_name: Optional[str], class_name: str):
def create_object(self, object_name: Optional[str], class_name: str) -> UUID:
obj = self.od.create_object(object_name, class_name)
self.__recompute_mappings()
self.recompute_mappings()
return obj
# internal use
@ -284,6 +279,6 @@ def bind_api(odapi):
'create_object': odapi.create_object,
'create_link': odapi.create_link,
'delete': odapi.delete,
'set_slot_value': odapi.set_slot_value,
'set_slot_value': odapi.set_slot_value
}
return funcs