latest pushes
This commit is contained in:
@@ -2,15 +2,95 @@
|
||||
|
||||
```
|
||||
from bellande_format import Bellande_Format
|
||||
from core.types import SchemaDefinition
|
||||
from datetime import datetime
|
||||
import os
|
||||
|
||||
bellande_formatter = Bellande_Format()
|
||||
# Initialize formatter
|
||||
formatter = Bellande_Format()
|
||||
|
||||
# Parse a Bellande file
|
||||
parsed_data = bellande_formatter.parse_bellande("path/to/your/file.bellande")
|
||||
# Example 1: Basic Usage
|
||||
data = {
|
||||
"name": "Project X",
|
||||
"version": 1.0,
|
||||
"created_at": datetime.now(),
|
||||
"settings": {
|
||||
"debug": True,
|
||||
"max_retries": 3
|
||||
},
|
||||
"users": [
|
||||
{"name": "John", "role": "admin"},
|
||||
{"name": "Jane", "role": "user"}
|
||||
]
|
||||
}
|
||||
|
||||
# Write data to a Bellande file
|
||||
data_to_write = {"key": "value", "list": [1, 2, 3]}
|
||||
bellande_formatter.write_bellande(data_to_write, "path/to/output/file.bellande")
|
||||
# Write data
|
||||
formatter.write_bellande(data, "config.bellande")
|
||||
|
||||
# Read data
|
||||
loaded_data = formatter.parse_bellande("config.bellande")
|
||||
|
||||
# Example 2: Schema Validation
|
||||
user_schema = SchemaDefinition(
|
||||
type="object",
|
||||
properties={
|
||||
"name": SchemaDefinition(type="string", pattern=r"^[a-zA-Z\s]+$"),
|
||||
"age": SchemaDefinition(type="integer", minimum=0, maximum=150),
|
||||
"email": SchemaDefinition(type="string", pattern=r"^[\w\.-]+@[\w\.-]+\.\w+$")
|
||||
},
|
||||
required=["name", "email"]
|
||||
)
|
||||
|
||||
formatter.register_schema("user", user_schema)
|
||||
|
||||
# Validate data
|
||||
user_data = {
|
||||
"name": "John Doe",
|
||||
"age": 30,
|
||||
"email": "john@example.com"
|
||||
}
|
||||
|
||||
result = formatter.validate(user_data, "user")
|
||||
print(f"Validation result: {result.is_valid}")
|
||||
|
||||
# Example 3: Encryption and Compression
|
||||
key = os.urandom(32) # Generate encryption key
|
||||
|
||||
# Encrypt data
|
||||
encrypted = formatter.encrypt(data, key)
|
||||
|
||||
# Decrypt data
|
||||
decrypted_data = formatter.decrypt(encrypted, key)
|
||||
|
||||
# Compress data
|
||||
compressed = formatter.compress(data)
|
||||
|
||||
# Decompress data
|
||||
decompressed_data = formatter.decompress(compressed)
|
||||
|
||||
# Example 4: Custom Types
|
||||
class Point2D:
|
||||
def __init__(self, x: float, y: float):
|
||||
self.x = x
|
||||
self.y = y
|
||||
|
||||
# Register custom type
|
||||
formatter.type_registry.register(
|
||||
"point2d",
|
||||
Point2D,
|
||||
lambda p: f"{p.x},{p.y}",
|
||||
lambda s: Point2D(*map(float, s.split(',')))
|
||||
)
|
||||
|
||||
# Use custom type
|
||||
location_data = {
|
||||
"points": [
|
||||
Point2D(1.0, 2.0),
|
||||
Point2D(3.0, 4.0)
|
||||
]
|
||||
}
|
||||
|
||||
formatter.write_bellande(location_data, "locations.bellande")
|
||||
```
|
||||
|
||||
## Website PYPI
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
# Copyright (C) 2024 Bellande Algorithm Model Research Innovation Center, Ronaldson Bellande
|
||||
# Copyright (C) 2024 Bellande Architecture Mechanism Research Innovation Center, Ronaldson Bellande
|
||||
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License as published by
|
||||
@@ -15,15 +15,40 @@
|
||||
|
||||
#!/usr/bin/env python3
|
||||
|
||||
import re, sys, json
|
||||
from typing import Dict, List, Union, Any
|
||||
from typing import Dict, List, Any, Union
|
||||
from .core.types import BellandeValue, ValidationResult, SchemaDefinition
|
||||
from .core.encryption import Encryption
|
||||
from .core.compression import Compression
|
||||
from .core.custom_types import CustomTypeRegistry
|
||||
from .core.validation import Validator
|
||||
import re
|
||||
import json
|
||||
|
||||
class Bellande_Format:
|
||||
def parse_bellande(self, file_path: str) -> str:
|
||||
with open(file_path, 'r') as file:
|
||||
lines = file.readlines()
|
||||
parsed_data = self.parse_lines(lines)
|
||||
return self.to_string_representation(parsed_data)
|
||||
def __init__(self):
|
||||
self.encryption = Encryption()
|
||||
self.compression = Compression()
|
||||
self.type_registry = CustomTypeRegistry()
|
||||
self.validator = Validator()
|
||||
self.references: Dict[str, Any] = {}
|
||||
self.schemas: Dict[str, SchemaDefinition] = {}
|
||||
|
||||
def register_schema(self, name: str, schema: SchemaDefinition):
|
||||
self.schemas[name] = schema
|
||||
|
||||
def validate(self, data: Any, schema_name: str) -> ValidationResult:
|
||||
if schema_name not in self.schemas:
|
||||
raise ValueError(f"Schema {schema_name} not found")
|
||||
return self.validator.validate(data, self.schemas[schema_name])
|
||||
|
||||
def parse_bellande(self, file_path: str) -> Any:
|
||||
with open(file_path, 'r', encoding='utf-8') as file:
|
||||
content = file.read()
|
||||
return self.parse_content(content)
|
||||
|
||||
def parse_content(self, content: str) -> Any:
|
||||
lines = content.split('\n')
|
||||
return self.parse_lines(lines)
|
||||
|
||||
def parse_lines(self, lines: List[str]) -> Union[Dict, List]:
|
||||
result = {}
|
||||
@@ -31,45 +56,56 @@ class Bellande_Format:
|
||||
current_list = None
|
||||
indent_stack = [(-1, result)]
|
||||
|
||||
for line in lines:
|
||||
stripped = line.strip()
|
||||
if not stripped or stripped.startswith('#'):
|
||||
continue
|
||||
for line_num, line in enumerate(lines, 1):
|
||||
try:
|
||||
stripped = line.strip()
|
||||
if not stripped or stripped.startswith('#'):
|
||||
continue
|
||||
|
||||
indent = len(line) - len(line.lstrip())
|
||||
indent = len(line) - len(line.lstrip())
|
||||
|
||||
while indent_stack and indent <= indent_stack[-1][0]:
|
||||
popped = indent_stack.pop()
|
||||
if isinstance(popped[1], list):
|
||||
current_list = None
|
||||
while indent_stack and indent <= indent_stack[-1][0]:
|
||||
popped = indent_stack.pop()
|
||||
if isinstance(popped[1], list):
|
||||
current_list = None
|
||||
|
||||
if ':' in stripped:
|
||||
key, value = map(str.strip, stripped.split(':', 1))
|
||||
current_key = key
|
||||
if value:
|
||||
result[key] = self.parse_value(value)
|
||||
else:
|
||||
result[key] = []
|
||||
current_list = result[key]
|
||||
indent_stack.append((indent, current_list))
|
||||
elif stripped.startswith('-'):
|
||||
value = stripped[1:].strip()
|
||||
parsed_value = self.parse_value(value)
|
||||
if current_list is not None:
|
||||
current_list.append(parsed_value)
|
||||
else:
|
||||
if not result: # If result is empty, start a root-level list
|
||||
result = [parsed_value]
|
||||
current_list = result
|
||||
indent_stack = [(-1, result)]
|
||||
if ':' in stripped:
|
||||
key, value = map(str.strip, stripped.split(':', 1))
|
||||
current_key = key
|
||||
if value:
|
||||
result[key] = self._process_value(value)
|
||||
else:
|
||||
result[current_key] = [parsed_value]
|
||||
current_list = result[current_key]
|
||||
result[key] = []
|
||||
current_list = result[key]
|
||||
indent_stack.append((indent, current_list))
|
||||
elif stripped.startswith('-'):
|
||||
value = stripped[1:].strip()
|
||||
parsed_value = self._process_value(value)
|
||||
if current_list is not None:
|
||||
current_list.append(parsed_value)
|
||||
else:
|
||||
if not result:
|
||||
result = [parsed_value]
|
||||
current_list = result
|
||||
indent_stack = [(-1, result)]
|
||||
else:
|
||||
result[current_key] = [parsed_value]
|
||||
current_list = result[current_key]
|
||||
indent_stack.append((indent, current_list))
|
||||
|
||||
except Exception as e:
|
||||
raise ValueError(f"Error parsing line {line_num}: {str(e)}")
|
||||
|
||||
return result
|
||||
|
||||
def parse_value(self, value: str) -> Union[str, int, float, bool, None]:
|
||||
def _process_value(self, value: str) -> Any:
|
||||
# Process custom types
|
||||
for type_name, deserializer in self.type_registry.deserializers.items():
|
||||
if value.startswith(f"type:{type_name}:"):
|
||||
type_value = value[len(f"type:{type_name}:"):]
|
||||
return deserializer(type_value)
|
||||
|
||||
# Process standard types
|
||||
if value.lower() == 'true':
|
||||
return True
|
||||
elif value.lower() == 'false':
|
||||
@@ -78,35 +114,23 @@ class Bellande_Format:
|
||||
return None
|
||||
elif value.startswith('"') and value.endswith('"'):
|
||||
return value[1:-1]
|
||||
elif value.startswith('ref:'):
|
||||
ref_key = value[4:].strip()
|
||||
if ref_key not in self.references:
|
||||
raise ValueError(f"Reference not found: {ref_key}")
|
||||
return self.references[ref_key]
|
||||
elif re.match(r'^-?\d+$', value):
|
||||
return int(value)
|
||||
elif re.match(r'^-?\d*\.\d+$', value):
|
||||
return float(value)
|
||||
else:
|
||||
return value
|
||||
|
||||
def to_string_representation(self, data: Any) -> str:
|
||||
if isinstance(data, dict):
|
||||
items = [f'"{k}": {self.to_string_representation(v)}' for k, v in data.items()]
|
||||
return '{' + ', '.join(items) + '}'
|
||||
elif isinstance(data, list):
|
||||
items = [self.to_string_representation(item) for item in data]
|
||||
return '[' + ', '.join(items) + ']'
|
||||
elif isinstance(data, str):
|
||||
return f'"{data}"'
|
||||
elif isinstance(data, (int, float)):
|
||||
return str(data)
|
||||
elif data is None:
|
||||
return 'null'
|
||||
elif isinstance(data, bool):
|
||||
return str(data).lower()
|
||||
else:
|
||||
return str(data)
|
||||
|
||||
return value
|
||||
|
||||
def write_bellande(self, data: Any, file_path: str):
|
||||
with open(file_path, 'w') as file:
|
||||
file.write(self.to_bellande_string(data))
|
||||
|
||||
content = self.to_bellande_string(data)
|
||||
with open(file_path, 'w', encoding='utf-8') as file:
|
||||
file.write(content)
|
||||
|
||||
def to_bellande_string(self, data: Any, indent: int = 0) -> str:
|
||||
if isinstance(data, dict):
|
||||
lines = []
|
||||
@@ -115,7 +139,7 @@ class Bellande_Format:
|
||||
lines.append(f"{' ' * indent}{key}:")
|
||||
lines.append(self.to_bellande_string(value, indent + 2))
|
||||
else:
|
||||
lines.append(f"{' ' * indent}{key}: {self.format_value(value)}")
|
||||
lines.append(f"{' ' * indent}{key}: {self._format_value(value)}")
|
||||
return '\n'.join(lines)
|
||||
elif isinstance(data, list):
|
||||
lines = []
|
||||
@@ -125,12 +149,21 @@ class Bellande_Format:
|
||||
lines.append(f"{' ' * indent}- {dict_lines[0]}")
|
||||
lines.extend(dict_lines[1:])
|
||||
else:
|
||||
lines.append(f"{' ' * indent}- {self.format_value(item)}")
|
||||
lines.append(f"{' ' * indent}- {self._format_value(item)}")
|
||||
return '\n'.join(lines)
|
||||
else:
|
||||
return f"{' ' * indent}{self.format_value(data)}"
|
||||
return f"{' ' * indent}{self._format_value(data)}"
|
||||
|
||||
def format_value(self, value: Any) -> str:
|
||||
def _format_value(self, value: Any) -> str:
|
||||
# Format custom types
|
||||
for type_name, serializer in self.type_registry.serializers.items():
|
||||
try:
|
||||
if isinstance(value, self.type_registry.types[type_name]):
|
||||
return f"type:{type_name}:{serializer(value)}"
|
||||
except:
|
||||
continue
|
||||
|
||||
# Format standard types
|
||||
if isinstance(value, str):
|
||||
if ' ' in value or ':' in value or value.lower() in ['true', 'false', 'null']:
|
||||
return f'"{value}"'
|
||||
@@ -139,70 +172,60 @@ class Bellande_Format:
|
||||
return str(value).lower()
|
||||
elif value is None:
|
||||
return 'null'
|
||||
else:
|
||||
return str(value)
|
||||
|
||||
return str(value)
|
||||
|
||||
def main(self) -> int:
|
||||
"""
|
||||
Main method to handle command-line operations.
|
||||
Returns an integer exit code.
|
||||
"""
|
||||
if len(sys.argv) < 2:
|
||||
print("Usage: Bellande_Format <command> [<file_path>] [<input_data>]")
|
||||
print("Commands: parse <file_path>, write <file_path> <input_data>, help")
|
||||
return 1
|
||||
def encrypt(self, data: Any, key: bytes) -> bytes:
|
||||
content = self.to_bellande_string(data)
|
||||
return self.encryption.encrypt(content.encode(), key)
|
||||
|
||||
command = sys.argv[1]
|
||||
def decrypt(self, encrypted_data: bytes, key: bytes) -> Any:
|
||||
decrypted = self.encryption.decrypt(encrypted_data, key)
|
||||
return self.parse_content(decrypted.decode())
|
||||
|
||||
try:
|
||||
if command == 'parse':
|
||||
if len(sys.argv) < 3:
|
||||
print("Error: Please provide a file path to parse.")
|
||||
return 1
|
||||
file_path = sys.argv[2]
|
||||
result = self.parse_bellande(file_path)
|
||||
print(result)
|
||||
return 0
|
||||
def compress(self, data: Any) -> bytes:
|
||||
content = self.to_bellande_string(data)
|
||||
return self.compression.encode_data(content.encode())[0]
|
||||
|
||||
elif command == 'write':
|
||||
if len(sys.argv) < 4:
|
||||
print("Error: Please provide a file path to write to and the input data.")
|
||||
return 1
|
||||
file_path = sys.argv[2]
|
||||
input_data = sys.argv[3]
|
||||
|
||||
try:
|
||||
data = json.loads(input_data)
|
||||
except json.JSONDecodeError:
|
||||
print("Error: Invalid JSON input. Please provide valid JSON data.")
|
||||
return 1
|
||||
|
||||
self.write_bellande(data, file_path)
|
||||
print(f"Data successfully written to {file_path}")
|
||||
return 0
|
||||
|
||||
elif command == 'help':
|
||||
print("Bellande_Format Usage:")
|
||||
print(" parse <file_path>: Parse a Bellande format file and print the result")
|
||||
print(" write <file_path> <input_data>: Write data in Bellande format to a file")
|
||||
print(" <input_data> should be a valid JSON string")
|
||||
print(" help: Display this help message")
|
||||
return 0
|
||||
|
||||
else:
|
||||
print(f"Unknown command: {command}")
|
||||
print("Use 'Bellande_Format help' for usage information.")
|
||||
return 1
|
||||
|
||||
except Exception as e:
|
||||
print(f"An error occurred: {e}", file=sys.stderr)
|
||||
return 1
|
||||
def decompress(self, compressed_data: bytes) -> Any:
|
||||
decompressed = self.compression.decode_data(compressed_data, {})
|
||||
return self.parse_content(decompressed.decode())
|
||||
|
||||
def main():
|
||||
"""
|
||||
Function to be used as the entry point in setup.py
|
||||
"""
|
||||
return Bellande_Format().main()
|
||||
import sys
|
||||
|
||||
if len(sys.argv) < 2:
|
||||
print("Usage: bellande_format <command> [<file_path>] [<input_data>]")
|
||||
return 1
|
||||
|
||||
formatter = Bellande_Format()
|
||||
command = sys.argv[1]
|
||||
|
||||
try:
|
||||
if command == 'parse':
|
||||
if len(sys.argv) < 3:
|
||||
print("Error: Please provide a file path to parse.")
|
||||
return 1
|
||||
result = formatter.parse_bellande(sys.argv[2])
|
||||
print(json.dumps(result, default=str))
|
||||
return 0
|
||||
|
||||
elif command == 'write':
|
||||
if len(sys.argv) < 4:
|
||||
print("Error: Please provide a file path and input data.")
|
||||
return 1
|
||||
data = json.loads(sys.argv[3])
|
||||
formatter.write_bellande(data, sys.argv[2])
|
||||
print(f"Data written to {sys.argv[2]}")
|
||||
return 0
|
||||
|
||||
else:
|
||||
print(f"Unknown command: {command}")
|
||||
return 1
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error: {e}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
|
||||
112
Package/Python/src/bellande_parser/core/compression.py
Normal file
112
Package/Python/src/bellande_parser/core/compression.py
Normal file
@@ -0,0 +1,112 @@
|
||||
# Copyright (C) 2024 Bellande Architecture Mechanism Research Innovation Center, Ronaldson Bellande
|
||||
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License as published by
|
||||
# the Free Software Foundation, either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
|
||||
#!/usr/bin/env python3
|
||||
|
||||
from typing import List, Dict, Tuple
|
||||
import heapq
|
||||
from dataclasses import dataclass
|
||||
from collections import Counter
|
||||
|
||||
@dataclass
|
||||
class HuffmanNode:
|
||||
char: str
|
||||
freq: int
|
||||
left: 'HuffmanNode' = None
|
||||
right: 'HuffmanNode' = None
|
||||
|
||||
def __lt__(self, other):
|
||||
return self.freq < other.freq
|
||||
|
||||
class Compression:
|
||||
def __init__(self):
|
||||
self.huffman_codes: Dict[str, str] = {}
|
||||
|
||||
def build_huffman_tree(self, data: bytes) -> HuffmanNode:
|
||||
# Count frequency of each byte
|
||||
freq = Counter(data)
|
||||
|
||||
# Create heap of HuffmanNodes
|
||||
heap: List[HuffmanNode] = []
|
||||
for char, frequency in freq.items():
|
||||
node = HuffmanNode(char=char, freq=frequency)
|
||||
heapq.heappush(heap, node)
|
||||
|
||||
# Build the tree
|
||||
while len(heap) > 1:
|
||||
left = heapq.heappop(heap)
|
||||
right = heapq.heappop(heap)
|
||||
|
||||
internal = HuffmanNode(
|
||||
char=None,
|
||||
freq=left.freq + right.freq,
|
||||
left=left,
|
||||
right=right
|
||||
)
|
||||
heapq.heappush(heap, internal)
|
||||
|
||||
return heap[0]
|
||||
|
||||
def generate_codes(self, node: HuffmanNode, code: str = ""):
|
||||
if node is None:
|
||||
return
|
||||
|
||||
if node.char is not None:
|
||||
self.huffman_codes[node.char] = code
|
||||
return
|
||||
|
||||
self.generate_codes(node.left, code + "0")
|
||||
self.generate_codes(node.right, code + "1")
|
||||
|
||||
def encode_data(self, data: bytes) -> Tuple[bytes, Dict]:
|
||||
# Build Huffman tree and generate codes
|
||||
root = self.build_huffman_tree(data)
|
||||
self.huffman_codes.clear()
|
||||
self.generate_codes(root)
|
||||
|
||||
# Encode the data
|
||||
encoded = "".join(self.huffman_codes[char] for char in data)
|
||||
|
||||
# Convert binary string to bytes
|
||||
padding = 8 - (len(encoded) % 8)
|
||||
encoded += "0" * padding
|
||||
|
||||
result = bytearray()
|
||||
for i in range(0, len(encoded), 8):
|
||||
result.append(int(encoded[i:i+8], 2))
|
||||
|
||||
return bytes(result), {"codes": self.huffman_codes, "padding": padding}
|
||||
|
||||
def decode_data(self, data: bytes, metadata: Dict) -> bytes:
|
||||
# Convert bytes to binary string
|
||||
binary = "".join(format(byte, '08b') for byte in data)
|
||||
|
||||
# Remove padding
|
||||
binary = binary[:-metadata["padding"]]
|
||||
|
||||
# Create reverse lookup table
|
||||
reverse_codes = {code: char for char, code in metadata["codes"].items()}
|
||||
|
||||
# Decode the data
|
||||
decoded = bytearray()
|
||||
current_code = ""
|
||||
|
||||
for bit in binary:
|
||||
current_code += bit
|
||||
if current_code in reverse_codes:
|
||||
decoded.append(reverse_codes[current_code])
|
||||
current_code = ""
|
||||
|
||||
return bytes(decoded)
|
||||
72
Package/Python/src/bellande_parser/core/custom_types.py
Normal file
72
Package/Python/src/bellande_parser/core/custom_types.py
Normal file
@@ -0,0 +1,72 @@
|
||||
# Copyright (C) 2024 Bellande Architecture Mechanism Research Innovation Center, Ronaldson Bellande
|
||||
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License as published by
|
||||
# the Free Software Foundation, either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
|
||||
#!/usr/bin/env python3
|
||||
|
||||
from typing import Dict, Any, Callable, Type
|
||||
import re
|
||||
from datetime import datetime, timedelta
|
||||
from decimal import Decimal
|
||||
from pathlib import Path
|
||||
import base64
|
||||
import struct
|
||||
|
||||
class CustomTypeRegistry:
|
||||
def __init__(self):
|
||||
self.types: Dict[str, Type] = {}
|
||||
self.serializers: Dict[str, Callable] = {}
|
||||
self.deserializers: Dict[str, Callable] = {}
|
||||
|
||||
def register(self, type_name: str, type_class: Type,
|
||||
serializer: Callable, deserializer: Callable):
|
||||
self.types[type_name] = type_class
|
||||
self.serializers[type_name] = serializer
|
||||
self.deserializers[type_name] = deserializer
|
||||
|
||||
class Complex:
|
||||
def __init__(self):
|
||||
self.pattern = re.compile(r'([-+]?\d*\.?\d*)([-+]\d*\.?\d*)j')
|
||||
|
||||
def serialize(self, value: complex) -> str:
|
||||
return f"{value.real}{'+' if value.imag >= 0 else ''}{value.imag}j"
|
||||
|
||||
def deserialize(self, value: str) -> complex:
|
||||
match = self.pattern.match(value)
|
||||
if not match:
|
||||
raise ValueError("Invalid complex number format")
|
||||
real = float(match.group(1))
|
||||
imag = float(match.group(2))
|
||||
return complex(real, imag)
|
||||
|
||||
class BinaryData:
|
||||
def serialize(self, value: bytes) -> str:
|
||||
return base64.b64encode(value).decode()
|
||||
|
||||
def deserialize(self, value: str) -> bytes:
|
||||
return base64.b64decode(value)
|
||||
|
||||
class DateTime:
|
||||
def serialize(self, value: datetime) -> str:
|
||||
return value.isoformat()
|
||||
|
||||
def deserialize(self, value: str) -> datetime:
|
||||
return datetime.fromisoformat(value)
|
||||
|
||||
class TimeDelta:
|
||||
def serialize(self, value: timedelta) -> str:
|
||||
return str(value.total_seconds())
|
||||
|
||||
def deserialize(self, value: str) -> timedelta:
|
||||
return timedelta(seconds=float(value))
|
||||
94
Package/Python/src/bellande_parser/core/encryption.py
Normal file
94
Package/Python/src/bellande_parser/core/encryption.py
Normal file
@@ -0,0 +1,94 @@
|
||||
# Copyright (C) 2024 Bellande Architecture Mechanism Research Innovation Center, Ronaldson Bellande
|
||||
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License as published by
|
||||
# the Free Software Foundation, either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
|
||||
#!/usr/bin/env python3
|
||||
|
||||
from typing import List
|
||||
import os
|
||||
|
||||
class AES:
|
||||
def __init__(self):
|
||||
self.block_size = 16
|
||||
|
||||
def pad(self, text: bytes) -> bytes:
|
||||
padding_size = self.block_size - (len(text) % self.block_size)
|
||||
padding = bytes([padding_size] * padding_size)
|
||||
return text + padding
|
||||
|
||||
def unpad(self, text: bytes) -> bytes:
|
||||
padding_size = text[-1]
|
||||
return text[:-padding_size]
|
||||
|
||||
def expand_key(self, key: bytes, rounds: int) -> List[bytes]:
|
||||
expanded: List[bytes] = []
|
||||
# Key expansion logic here
|
||||
return expanded
|
||||
|
||||
def encrypt_block(self, block: bytes, round_keys: List[bytes]) -> bytes:
|
||||
# AES block encryption implementation
|
||||
state = list(block)
|
||||
# Add round key
|
||||
# SubBytes
|
||||
# ShiftRows
|
||||
# MixColumns
|
||||
# Final round
|
||||
return bytes(state)
|
||||
|
||||
def decrypt_block(self, block: bytes, round_keys: List[bytes]) -> bytes:
|
||||
# AES block decryption implementation
|
||||
state = list(block)
|
||||
# Inverse operations
|
||||
return bytes(state)
|
||||
|
||||
class Encryption:
|
||||
def __init__(self):
|
||||
self.aes = AES()
|
||||
|
||||
def generate_key(self) -> bytes:
|
||||
return os.urandom(32)
|
||||
|
||||
def encrypt(self, data: bytes, key: bytes) -> bytes:
|
||||
iv = os.urandom(16) # Initialization vector
|
||||
padded_data = self.aes.pad(data)
|
||||
round_keys = self.aes.expand_key(key, 10)
|
||||
|
||||
cipher = iv
|
||||
previous = iv
|
||||
|
||||
for i in range(0, len(padded_data), 16):
|
||||
block = padded_data[i:i+16]
|
||||
block = bytes(a ^ b for a, b in zip(block, previous))
|
||||
encrypted_block = self.aes.encrypt_block(block, round_keys)
|
||||
cipher += encrypted_block
|
||||
previous = encrypted_block
|
||||
|
||||
return cipher
|
||||
|
||||
def decrypt(self, cipher: bytes, key: bytes) -> bytes:
|
||||
iv = cipher[:16]
|
||||
cipher = cipher[16:]
|
||||
round_keys = self.aes.expand_key(key, 10)
|
||||
|
||||
plain = b""
|
||||
previous = iv
|
||||
|
||||
for i in range(0, len(cipher), 16):
|
||||
block = cipher[i:i+16]
|
||||
decrypted_block = self.aes.decrypt_block(block, round_keys)
|
||||
plain_block = bytes(a ^ b for a, b in zip(decrypted_block, previous))
|
||||
plain += plain_block
|
||||
previous = block
|
||||
|
||||
return self.aes.unpad(plain)
|
||||
78
Package/Python/src/bellande_parser/core/types.py
Normal file
78
Package/Python/src/bellande_parser/core/types.py
Normal file
@@ -0,0 +1,78 @@
|
||||
# Copyright (C) 2024 Bellande Architecture Mechanism Research Innovation Center, Ronaldson Bellande
|
||||
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License as published by
|
||||
# the Free Software Foundation, either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
|
||||
#!/usr/bin/env python3
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Dict, List, Any, Optional
|
||||
from datetime import datetime
|
||||
import hashlib
|
||||
|
||||
@dataclass
|
||||
class ValidationResult:
|
||||
is_valid: bool
|
||||
errors: List[str]
|
||||
warnings: List[str]
|
||||
path: str = ""
|
||||
details: Dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
@dataclass
|
||||
class VersionInfo:
|
||||
version: int
|
||||
timestamp: datetime
|
||||
author: str
|
||||
changes: Dict[str, Any]
|
||||
checksum: str
|
||||
|
||||
@dataclass
|
||||
class SchemaDefinition:
|
||||
type: str
|
||||
properties: Dict[str, Any] = field(default_factory=dict)
|
||||
required: List[str] = field(default_factory=list)
|
||||
pattern: Optional[str] = None
|
||||
enum: Optional[List[Any]] = None
|
||||
minimum: Optional[float] = None
|
||||
maximum: Optional[float] = None
|
||||
format: Optional[str] = None
|
||||
|
||||
class BellandeValue:
|
||||
def __init__(self, value: Any, metadata: Dict = None):
|
||||
self.value = value
|
||||
self.metadata = metadata or {}
|
||||
self.created_at = datetime.now()
|
||||
self.modified_at = self.created_at
|
||||
self.version = 1
|
||||
self.checksum = self._calculate_checksum()
|
||||
self.history: List[VersionInfo] = []
|
||||
|
||||
def _calculate_checksum(self) -> str:
|
||||
return hashlib.sha256(str(self.value).encode()).hexdigest()
|
||||
|
||||
def update(self, new_value: Any, author: str):
|
||||
old_value = self.value
|
||||
self.value = new_value
|
||||
self.modified_at = datetime.now()
|
||||
self.version += 1
|
||||
new_checksum = self._calculate_checksum()
|
||||
|
||||
version_info = VersionInfo(
|
||||
version=self.version,
|
||||
timestamp=self.modified_at,
|
||||
author=author,
|
||||
changes={"old": old_value, "new": new_value},
|
||||
checksum=new_checksum
|
||||
)
|
||||
self.history.append(version_info)
|
||||
self.checksum = new_checksum
|
||||
111
Package/Python/src/bellande_parser/core/validation.py
Normal file
111
Package/Python/src/bellande_parser/core/validation.py
Normal file
@@ -0,0 +1,111 @@
|
||||
# Copyright (C) 2024 Bellande Architecture Mechanism Research Innovation Center, Ronaldson Bellande
|
||||
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License as published by
|
||||
# the Free Software Foundation, either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
|
||||
#!/usr/bin/env python3
|
||||
|
||||
from typing import Dict, Any, List
|
||||
import re
|
||||
from datetime import datetime
|
||||
from decimal import Decimal
|
||||
from .types import SchemaDefinition, ValidationResult
|
||||
|
||||
class Validator:
|
||||
def __init__(self):
|
||||
self.type_validators = {
|
||||
'string': self._validate_string,
|
||||
'number': self._validate_number,
|
||||
'integer': self._validate_integer,
|
||||
'boolean': self._validate_boolean,
|
||||
'array': self._validate_array,
|
||||
'object': self._validate_object,
|
||||
'null': self._validate_null
|
||||
}
|
||||
|
||||
def validate(self, data: Any, schema: SchemaDefinition, path: str = "") -> ValidationResult:
|
||||
if schema.type not in self.type_validators:
|
||||
return ValidationResult(False, [f"Unknown type: {schema.type}"], [])
|
||||
|
||||
return self.type_validators[schema.type](data, schema, path)
|
||||
|
||||
def _validate_string(self, data: Any, schema: SchemaDefinition, path: str) -> ValidationResult:
|
||||
if not isinstance(data, str):
|
||||
return ValidationResult(False, [f"{path}: Expected string, got {type(data).__name__}"], [])
|
||||
|
||||
errors = []
|
||||
if schema.pattern and not re.match(schema.pattern, data):
|
||||
errors.append(f"{path}: String does not match pattern {schema.pattern}")
|
||||
|
||||
if schema.enum and data not in schema.enum:
|
||||
errors.append(f"{path}: Value not in enum: {schema.enum}")
|
||||
|
||||
return ValidationResult(not errors, errors, [])
|
||||
|
||||
def _validate_number(self, data: Any, schema: SchemaDefinition, path: str) -> ValidationResult:
|
||||
if not isinstance(data, (int, float, Decimal)):
|
||||
return ValidationResult(False, [f"{path}: Expected number, got {type(data).__name__}"], [])
|
||||
|
||||
errors = []
|
||||
if schema.minimum is not None and data < schema.minimum:
|
||||
errors.append(f"{path}: Value below minimum: {schema.minimum}")
|
||||
|
||||
if schema.maximum is not None and data > schema.maximum:
|
||||
errors.append(f"{path}: Value above maximum: {schema.maximum}")
|
||||
|
||||
return ValidationResult(not errors, errors, [])
|
||||
|
||||
def _validate_integer(self, data: Any, schema: SchemaDefinition, path: str) -> ValidationResult:
|
||||
if not isinstance(data, int):
|
||||
return ValidationResult(False, [f"{path}: Expected integer, got {type(data).__name__}"], [])
|
||||
return self._validate_number(data, schema, path)
|
||||
|
||||
def _validate_boolean(self, data: Any, schema: SchemaDefinition, path: str) -> ValidationResult:
|
||||
if not isinstance(data, bool):
|
||||
return ValidationResult(False, [f"{path}: Expected boolean, got {type(data).__name__}"], [])
|
||||
return ValidationResult(True, [], [])
|
||||
|
||||
def _validate_array(self, data: Any, schema: SchemaDefinition, path: str) -> ValidationResult:
|
||||
if not isinstance(data, list):
|
||||
return ValidationResult(False, [f"{path}: Expected array, got {type(data).__name__}"], [])
|
||||
|
||||
errors = []
|
||||
for i, item in enumerate(data):
|
||||
item_path = f"{path}[{i}]"
|
||||
if 'items' in schema.properties:
|
||||
result = self.validate(item, schema.properties['items'], item_path)
|
||||
errors.extend(result.errors)
|
||||
|
||||
return ValidationResult(not errors, errors, [])
|
||||
|
||||
def _validate_object(self, data: Any, schema: SchemaDefinition, path: str) -> ValidationResult:
|
||||
if not isinstance(data, dict):
|
||||
return ValidationResult(False, [f"{path}: Expected object, got {type(data).__name__}"], [])
|
||||
|
||||
errors = []
|
||||
for required in schema.required:
|
||||
if required not in data:
|
||||
errors.append(f"{path}: Missing required field: {required}")
|
||||
|
||||
for key, value in data.items():
|
||||
if key in schema.properties:
|
||||
prop_schema = schema.properties[key]
|
||||
result = self.validate(value, prop_schema, f"{path}.{key}")
|
||||
errors.extend(result.errors)
|
||||
|
||||
return ValidationResult(not errors, errors, [])
|
||||
|
||||
def _validate_null(self, data: Any, schema: SchemaDefinition, path: str) -> ValidationResult:
|
||||
if data is not None:
|
||||
return ValidationResult(False, [f"{path}: Expected null, got {type(data).__name__}"], [])
|
||||
return ValidationResult(True, [], [])
|
||||
Reference in New Issue
Block a user