Skip to content

Commit

Permalink
Information methods and Vector Types Constants
Browse files Browse the repository at this point in the history 
- The ``dict()`` method is added to the ``Information`` class to facilitate the handling of ARP files.

- ``Vector`` types have been declared as constants to improve code handling, scalability, and readability.

- The way the names of environments and situations are generated has been changed, correcting a security issue.

- Improved code readability and style.
  • Loading branch information
@aamat-ausa
aamat-ausa committed May 7, 2024
1 parent c050244 commit 3a64e5b
Show file tree
Hide file tree
Showing 18 changed files with 162 additions and 572 deletions.
17 changes: 16 additions & 1 deletion AkitaCode/__init__.py
View file
Original file line number Diff line number Diff line change
@@ -1 +1,16 @@
from . import bd, canbus, conditionals, data, document, frame, information, instances, keys, line_state_machine, line, mask, messages, numerics, protocol, situation
from . import bd
from . import canbus
from . import conditionals
from . import data
from . import document
from . import frame
from . import information
from . import instances
from . import keys
from . import line_state_machine
from . import line
from . import mask
from . import messages
from . import numerics
from . import protocol
from . import situation
136 changes: 1 addition & 135 deletions AkitaCode/bd.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2776,138 +2776,4 @@ def export_all_as_json(self):
return json.dumps(full_db)

except Exception as e:
raise Exception(f"No s'ha pogut exportar la base de dades. Error: {e}")

if __name__ == "__main__":
#from config_mod import *
db = Database(r"C:\Development\Akitacan Code\new_proves.db")
#print(db.db)
if db.loaded_db():
#print(db.get_all_used_msg_id_from_protocol_except_f_id(1,3))
#print(db.modify_function(4, "TEST2", "es un test2", "0x300", 6))
"""
print("------------------------------------------FIRST TEST------------------------------------------")
db.add_protocol("T164E","Aquest protocol es per el telescopi electric d'Ausa")
db.add_variable("BAT_Number_of_Batteries", "", 0, "301", bytes([0,0,0,0,0,0,0,255]), 0, "0", 0, 0, 0, 1)
db.add_variable("BAT_VIRTUAL_BATTERY_SOC", "", 0, "301", bytes([0,0,0,0,0,0,255,0]), 0, "0", 0, 0, 0, 1)
db.add_variable("BAT_VIRTUAL_BATTERY_TEMPERATURE", "", 0, "303", bytes([0,0,0,0,0,0,255,255]), 0, "0", 0, 0, 0, 1)
db.add_function("SDO_TEST", "", "631", 1 )
db.add_argument("B0", "", 0, bytes([0,0,0,0,0,0,0,255]), 0, 0, 0, 3)
db.add_argument("B1", "", 0, bytes([0,0,0,0,0,0,255,0]), 0, 0, 0, 3)
print("------------------------------------------SECOND TEST------------------------------------------")
print(db.get_variables_from_protocol(1))
print(db.get_functions_from_protocol(1))
print(db.get_arguments_from_function(3))
print("------------------------------------------THIRD TEST------------------------------------------")
print(db.get_nbits_from_variable(1))
print(db.get_nbits_from_variable(3))
print(db.get_nbits_from_argument(3))
print(db.get_protocol_id("T164E"))
print(db.get_protocol_name(1))
print(db.get_default_variable_value(1,"BAT_NUMBER_OF_BATTERIES"))
print(db.exist_any_protocol())
print(db.update_variable_mask(1, 0xff0000))
print(db.update_argument_mask(3, 0xff0000))
print(db.get_variable_mask(3))
print(db.get_argument_mask(3))
print("------------------------------------------FORTH TEST------------------------------------------")
print(db.get_variable_id(1,"BAT_NUMBER_OF_BATTERIES"))
print(db.get_function_id(1,"SDO_TEST"))
print(db.get_argument_id(3,"B1"))
print(db.get_variable_name(2))
print(db.get_function_name(3))
print(db.get_argument_name(3))
print(db.get_variable_canid(1))
print(db.get_function_canid(3))
print(db.get_variable_is_signed(1))
print(db.get_argument_is_signed(3))
print(db.get_variable_direction(1))
print(db.get_all_protocols_names())
print(db.get_all_variables_masks_from_protocol(1,769,0))
print(db.get_all_variables_masks_from_protocol_except_v_id(1,1,0))
print(db.get_all_argument_masks_from_function(3))
print(db.get_all_argument_masks_from_function_except_a_id(3,3))
print("------------------------------------------FIFTH TEST------------------------------------------")
print(db.get_all_variables_names_from_protocol(1))
print(db.get_all_variables_names_from_protocol_except_v_id(1,1))
print(db.get_all_arguments_names_from_function_except_a_id(3,4))
print(db.get_all_from_variable(1))
print(db.get_all_from_function(3))
print(db.get_all_from_argument(3))
print(db.get_all_variables())
print(db.get_all_functions())
print(db.get_all_arguments())
print(db.get_all_structured_by_p_id())
print("------------------------------------------SIXTH TEST------------------------------------------")
print(db.get_info_from_protocol(1))
#print(db.modify_protocol(1,"T164E","Aquest protocol es per el telescopi electric d'Ausa")) #FUNCIONA PERO ES COMENTA X EVITAR PROBLEMES
#print(db.delete_protocol(3)) #FUNCIONA PERO ES COMENTA X EVITAR PROBLEMES
print(db.get_info_from_variable(1))
#print(db.modify_variable(1,"BAT_NUMBER_OF_BATTERIES","",0,"301",bytes([0,0,0,0,0,255,0,0]),0,0,0,0,0,1)) #FUNCIONA PERO ES COMENTA X EVITAR PROBLEMES
print(db.get_specific_p_id_from_v_id(3))
#print(db.delete_variable(6)) #FUNCIONA PERO ES COMENTA X EVITAR PROBLEMES
print(db.get_info_from_function(3))
print(db.get_all_used_msg_id_from_protocol(1))
print(db.get_all_functions_names_from_protocol_except_f_id(1,3))
#print(db.modify_function(3,"SDO", "", "631", 1)) #FUNCIONA PERO ES COMENTA X EVITAR PROBLEMES
#print(db.delete_function(4)) #FUNCIONA PERO ES COMENTA X EVITAR PROBLEMES
#print(db.delete_argument(5)) #FUNCIONA PERO ES COMENTA X EVITAR PROBLEMES
print(db.get_specific_p_id_from_f_id(3))
print("------------------------------------------SEVENTH TEST------------------------------------------")
print(db.extract_protocol(1)) #VA PERO NO SE SI FA EL QUE ES VOLIA A LA IDEA ORIGINAL
#print(db.export_all_as_json())
#with open("my_db.json","w") as f:
#f.write(db.export_all_as_json())
print("------------------------------------------EIGHT TEST------------------------------------------")
VALIDAT
#PROTOCOL FUNCTIONS TO TEST
print("PROTOCOL TEST")
print(db.add_protocol("TEST","Aquest protocol es una prova"))
print(db.update_protocol_name(4,"PROTOCOL_TEST"))
print(db.update_protocol_description(4,"AIXO ES UN TEST"))
print(db.get_protocol_id("PROTOCOL_TEST"))
print(db.get_protocol_name(4))
print(db.get_protocol_description(4))
#VARIABLE FUNCTIONS TO TEST
print("VARIABLE TEST")
#print(db.add_variable("test","",1,"400",bytes([0,0,0,0,0,0,0,1]),0,0,0,0,0,4))
print(db.get_variable_description(7))
print(db.get_variable_direction(7))
print(db.get_variable_offset(7))
print(db.get_variable_mul(7))
print(db.get_variable_div(7))
print(db.update_variable_name(4,"TEST_FINAL"))
print(db.update_variable_description(7,"UNA DESCRIPCIO"))
print(db.update_variable_msg_id(7,"500"))
print(db.update_variable_direction(7,0))
print(db.update_variable_is_signed(7,1))
print(db.update_variable_default_value(7,1))
print(db.update_variable_offset(7,1))
print(db.update_variable_mul(7,1))
print(db.update_variable_div(7,1))
#FUNCTION FUNCTIONS TO TEST
print("FUNCTION TEST")
#print(db.add_function("TEST", "", "700", 1))
print(db.update_function_description(5,"DESCRIPCIO DE TEST"))
print(db.get_function_description(5))
print(db.update_function_name(1,5,"FUNCIO_TEST"))
print(db.update_function_name(1,3,"SDO_TEST"))
print(db.update_function_msg_id(5,"800"))
#ARGUMENT FUNCTIONS TO TEST
print("ARGUMENT TEST")
#print(db.add_argument("TEST","",0,bytes([0,0,0,0,0,0,0,255]),0,0,0,3))
print(db.update_argument_description(6,"descripcio"))
print(db.update_argument_name(3,6,"arg_test"))
print(db.update_argument_is_signed(6,1))
print(db.update_argument_offset(6,1))
print(db.update_argument_mul(6,1))
print(db.update_argument_div(6,1))
print(db.get_argument_description(6))
print(db.get_argument_offset(6))
print(db.get_argument_mul(6))
print(db.get_argument_div(6))
"""

raise Exception(f"No s'ha pogut exportar la base de dades. Error: {e}")
27 changes: 1 addition & 26 deletions AkitaCode/canbus.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -118,29 +118,4 @@ def cleanup(self):
# Atura el notifier que estem usant per llegir dades del bus
self.notifier.stop()
# Trenca la connexió del bus
self.bus.shutdown()


if __name__ == "__main__":
#pcan = Bus(filters=[{"can_id": 0x150, "can_mask": 0xFF9, "extended": False},])
pcan = Bus(bitrate=250000)
id_msg = 0x100
payload_msg = [1,2,3,4,5,6,7,8]
pcan.flush_buffer()
for e in list(range(0,100)):
if pcan.send_message(id_msg, payload_msg):
print("Success!")
else:
print("Failed!")
"""
try:
pcan.flush_buffer()
while(True):
msg_received = pcan.read_input()
if msg_received != None:
print(list(msg_received.data))
print(msg_received)
except:
logging.info("Sortint de l'aplicatiu.")
pcan.cleanup()
"""
self.bus.shutdown()
48 changes: 24 additions & 24 deletions AkitaCode/conditionals.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -72,29 +72,29 @@ def filtrar_combinaciones_mixtas(combinaciones, condiciones):
return combinaciones_filtradas

# Ejemplo de uso
if __name__ == "__main__":
import time
# from config_mod import get_db_path
# db = Database(get_db_path())
protocol = 1
db = "C:/Users/aamat/.akitacan/data.db"
# if __name__ == "__main__":
# import time
# # from config_mod import get_db_path
# # db = Database(get_db_path())
# protocol = 1
# db = "C:/Users/aamat/.akitacan/data.db"

variables = ["BAT_NUMBER_OF_BATTERIES", "BAT_VIRTUAL_BATTERY_SOC"]
# variables = ["BAT_VIRTUAL_BATTERY_SOC"]
condiciones = [
# ("A", "==", 1, None),
("BAT_NUMBER_OF_BATTERIES", "==", 0, "AND"),
("BAT_VIRTUAL_BATTERY_SOC", "==", 1, "OR"),
("BAT_NUMBER_OF_BATTERIES", "==", 1, "AND"),
("BAT_VIRTUAL_BATTERY_SOC", "==", 0, None)
]
init_time = time.time()
combinaciones = generate_combinations(db,protocol,variables,1)
print(combinaciones)
combinaciones_filtradas = filtrar_combinaciones_mixtas(combinaciones, condiciones)
print(combinaciones_filtradas)
print(time.time()-init_time)
with open("E:/prova_autogen.txt", "w") as f:
for e in combinaciones:
f.write(str(e)+"\n")
# variables = ["BAT_NUMBER_OF_BATTERIES", "BAT_VIRTUAL_BATTERY_SOC"]
# # variables = ["BAT_VIRTUAL_BATTERY_SOC"]
# condiciones = [
# # ("A", "==", 1, None),
# ("BAT_NUMBER_OF_BATTERIES", "==", 0, "AND"),
# ("BAT_VIRTUAL_BATTERY_SOC", "==", 1, "OR"),
# ("BAT_NUMBER_OF_BATTERIES", "==", 1, "AND"),
# ("BAT_VIRTUAL_BATTERY_SOC", "==", 0, None)
# ]
# init_time = time.time()
# combinaciones = generate_combinations(db,protocol,variables,1)
# print(combinaciones)
# combinaciones_filtradas = filtrar_combinaciones_mixtas(combinaciones, condiciones)
# print(combinaciones_filtradas)
# print(time.time()-init_time)
# with open("E:/prova_autogen.txt", "w") as f:
# for e in combinaciones:
# f.write(str(e)+"\n")

52 changes: 1 addition & 51 deletions AkitaCode/data.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -220,27 +220,6 @@ def __repr__(self) -> str:
:rtype: str
"""
return "Data Object: "+self.name+" takes up "+str(len(self.mask))+"bits."


# def get_all(self):
# """
# Retorna en format tupla i en el ordre següent la informació emmagatzemada a l'objecte Data.

# - El nom de la variable o funció.
# - El valor esperat abans de ser encapsulat.
# - Si la variable o funció és complement a dos o no.
# - El nombre de bits de la variable o funció en la trama.
# - L'objecte Mask de la variable o funció.
# - L'identificador de trama CAN.
# - Si la variable és de resposta o no. (Per les funcions, aquest paràmetre sempre és False)
# - Si l'objecte Data és una funció.
# - El valor de l'objecte Data un cop encapsulat.

# :return: Retorna tota l'informació de l'objecte Data.
# :rtype: tuple
# """
# return self.var_name, self.value, self.is_signed, self.n_bits, self.mask, self.can_id, self.is_response, self.is_function, self.frame_value



class Argument(Data):
Expand All @@ -263,33 +242,4 @@ def get_is_response(self) -> bool:
class Function(object):
def __init__(self,ids:int,name:str) -> None:
self.ids:int = ids
self.name:str = name
# self.arguments:list[Argument] = []






if __name__ == "__main__":
d = Data("HOLA", 184, 65280, 8, value=126, is_signed=True)
p = Data("ADEU", 587, 65283, 10, value=78, is_function=True)
print(d.get_name(), d.get_can_id(), d.get_frame_value(), d.get_is_function(), d.get_mask_value(), d.get_is_function()) # expected HOLA 184 32256 False b'\x00\x00\x00\x00\x00\x00\xff\x00' False
print(p.get_name(), p.get_can_id(), p.get_frame_value(), p.get_is_function(), p.get_mask_value(), p.get_is_function()) # expected ADEU 587 4866 True b'\x00\x00\x00\x00\x00\x00\xff\x03' True
# Test of uncapsulate()
a = Data("HOLA", 184, mask=65283, n_bits=10, value=-127, is_signed=True, is_response=True)
b = Data("ALOHA", 184, mask=250, n_bits=6, value=48, is_response=True)
print(a._uncapsulate((2**15)+3)) # expected -509
print(a._uncapsulate(57345)) # expected -127
# 1110000000000001 = 57345 // Value of received frame
# 1111111100000011 = 65283 // Mask of a Data object.
# 11100000------01 = ? // Execute uncapsulate.
# 1110000001 (signed) = -127 // Extracted value
print(b._uncapsulate(197)) # expected 48
# 11000101 = 197 // Value of received frame.
# 11111010 = 250 // Mask of b Data object.
# 11000-0- = ? // Execute uncapsulate.
# 110000 = 48 // Extracted value.
print(a._uncapsulate(57345) == a.get_expected_value())
print(b._uncapsulate(197) == b.get_expected_value())
print(d.get_value_is_signed())
self.name:str = name
Loading

0 comments on commit 3a64e5b

Please sign in to comment.