BRSRI/bellande_robotic_system_operating_architecture
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

recal

Browse Source
This commit is contained in:
Ronaldson Bellande
2025-03-20 11:21:43 -04:00
parent 42a586264b
commit 9297c26603
4 changed files with 597 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

87
src/brsoa_build.py Normal file
View File

@@ -0,0 +1,87 @@
# Copyright (C) 2025 Bellande Robotics Sensors 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/>.
import os
import subprocess
from bellande_parser.bellande_parser import Bellande_Format
def load_config():
bellande_parser = Bellande_Format()
raw_content = bellande_parser.parse_bellande("project_config.bellande")
def build_cpp(package_dir, output_dir, config):
src_file = os.path.join(package_dir, f"{os.path.basename(package_dir)}.cpp")
output_file = os.path.join(output_dir, os.path.basename(package_dir))
compiler = config['build_settings']['cpp']['compiler']
flags = ' '.join(config['build_settings']['cpp']['flags'])
cmd = f"{compiler} {flags} -I build/common_msgs {src_file} -o {output_file}"
subprocess.run(cmd, shell=True, check=True)
def build_python(package_dir, output_dir, config):
src_file = os.path.join(package_dir, f"{os.path.basename(package_dir)}.py")
output_file = os.path.join(output_dir, os.path.basename(src_file))
os.makedirs(output_dir, exist_ok=True)
os.symlink(src_file, output_file)
def build_java(package_dir, output_dir, config):
src_file = os.path.join(package_dir, f"{os.path.basename(package_dir)}.java")
output_dir = os.path.join(output_dir, "classes")
os.makedirs(output_dir, exist_ok=True)
cmd = f"javac -d {output_dir} -cp build/common_msgs {src_file}"
subprocess.run(cmd, shell=True, check=True)
def build_rust(package_dir, output_dir, config):
cmd = f"cargo build --release --manifest-path {os.path.join(package_dir, 'Cargo.toml')}"
subprocess.run(cmd, shell=True, check=True)
binary_name = os.path.basename(package_dir)
src = os.path.join(package_dir, "target", "release", binary_name)
dst = os.path.join(output_dir, binary_name)
os.makedirs(output_dir, exist_ok=True)
os.rename(src, dst)
def build_go(package_dir, output_dir, config):
cmd = f"go build -o {output_dir} {os.path.join(package_dir, '*.go')}"
subprocess.run(cmd, shell=True, check=True)
def main():
config = load_config()
# Generate common messages
subprocess.run(["python", "generate_msgs.py"], check=True)
build_functions = {
'cpp': build_cpp,
'python': build_python,
'java': build_java,
'rust': build_rust,
'go': build_go,
}
for package in config['packages']:
package_dir = os.path.join('src', package['name'])
output_dir = os.path.join('build', package['name'])
os.makedirs(output_dir, exist_ok=True)
lang = package['language']
if lang in build_functions:
build_functions[lang](package_dir, output_dir, config)
else:
print(f"Warning: No build function for language '{lang}'")
if __name__ == "__main__":
main()

187
src/brsoa_create_package.py Normal file
View File

@@ -0,0 +1,187 @@
# Copyright (C) 2025 Bellande Robotics Sensors 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/>.
import os
import argparse
from bellande_parser.bellande_parser import Bellande_Format
TEMPLATE_DIR = "templates"
def create_package_bellande(package_name, language, dependencies):
content = {
"name": package_name,
"language": language,
"dependencies": dependencies
}
return yaml.dump(content, default_flow_style=False)
def create_cpp_source(package_name):
return f"""#include "common_msgs.hpp"
#include <iostream>
#include <chrono>
#include <thread>
class {package_name.capitalize()} {{
public:
void run() {{
while (true) {{
// TODO: Implement node logic
std::cout << "Running {package_name}" << std::endl;
std::this_thread::sleep_for(std::chrono::seconds(1));
}}
}}
}};
int main(int argc, char** argv) {{
{package_name.capitalize()} node;
node.run();
return 0;
}}
"""
def create_python_source(package_name):
return f"""from common_msgs import *
import time
class {package_name.capitalize()}:
def run(self):
while True:
# TODO: Implement node logic
print(f"Running {package_name}")
time.sleep(1)
if __name__ == "__main__":
node = {package_name.capitalize()}()
node.run()
"""
def create_java_source(package_name):
return f"""import common_msgs.Messages.*;
public class {package_name.capitalize()} {{
public void run() {{
while (true) {{
// TODO: Implement node logic
System.out.println("Running {package_name}");
try {{
Thread.sleep(1000);
}} catch (InterruptedException e) {{
e.printStackTrace();
}}
}}
}}
public static void main(String[] args) {{
{package_name.capitalize()} node = new {package_name.capitalize()}();
node.run();
}}
}}
"""
def create_rust_source(package_name):
return f"""use common_msgs::*;
use std::{{thread, time}};
struct {package_name.capitalize()};
impl {package_name.capitalize()} {{
fn run(&self) {{
loop {{
// TODO: Implement node logic
println!("Running {package_name}");
thread::sleep(time::Duration::from_secs(1));
}}
}}
}}
fn main() {{
let node = {package_name.capitalize()}{{}};
node.run();
}}
"""
def create_go_source(package_name):
return f"""package main
import (
"fmt"
"time"
"common_msgs"
)
type {package_name.capitalize()} struct{{}}
func (n *{package_name.capitalize()}) Run() {{
for {{
// TODO: Implement node logic
fmt.Println("Running {package_name}")
time.Sleep(1 * time.Second)
}}
}}
func main() {{
node := &{package_name.capitalize()}{{}}
node.Run()
}}
"""
def create_package(package_name, language):
package_dir = os.path.join("src", package_name)
os.makedirs(package_dir, exist_ok=True)
# Create package.bellande
with open(os.path.join(package_dir, "package.bellande"), "w") as f:
f.write(create_package_bellande(package_name, language, ["common_msgs"]))
# Create source file
source_creators = {
"cpp": create_cpp_source,
"python": create_python_source,
"java": create_java_source,
"rust": create_rust_source,
"go": create_go_source
}
if language in source_creators:
source_content = source_creators[language](package_name)
source_filename = f"{package_name}.{language}"
if language == "cpp":
source_filename = f"{package_name}.cpp"
elif language == "python":
source_filename = f"{package_name}.py"
elif language == "java":
source_filename = f"{package_name.capitalize()}.java"
elif language == "rust":
source_filename = "main.rs"
os.makedirs(os.path.join(package_dir, "src"), exist_ok=True)
with open(os.path.join(package_dir, "Cargo.toml"), "w") as f:
f.write(f"[package]\nname = \"{package_name}\"\nversion = \"0.1.0\"\nedition = \"2021\"\n\n[dependencies]\ncommon_msgs = {{ path = \"../../build/common_msgs\" }}\n")
elif language == "go":
source_filename = f"{package_name}.go"
with open(os.path.join(package_dir, source_filename), "w") as f:
f.write(source_content)
else:
print(f"Unsupported language: {language}")
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Create a new package for the robot architecture.")
parser.add_argument("package_name", help="Name of the package to create")
parser.add_argument("language", choices=["cpp", "python", "java", "rust", "go"], help="Programming language for the package")
args = parser.parse_args()
create_package(args.package_name, args.language)
print(f"Created package {args.package_name} using {args.language}")

131
src/brsoa_create_system_config.py Normal file
View File

@@ -0,0 +1,131 @@
# Copyright (C) 2025 Bellande Robotics Sensors 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
# Will later be programming bellronos
def create_example_lua_config_file(filename="brsoa_system_config.lua"):
"""
Creates an example Lua configuration file for a robot system with generic settings.
Args:
filename (str): The name of the file to create
Returns:
bool: True if file creation was successful
"""
# Configuration content with generic example values
config_content = """-- brsoa_system_config.lua
-- Global configuration
global_config = {
max_nodes = 50,
discovery_method = "broadcast",
system_name = "example_system",
log_level = "debug"
}
-- Define nodes to be launched
nodes = {
{
name = "example_node_cpp",
package = "example_package_cpp",
executable = "example_driver_cpp",
language = "cpp",
args = {"--fps=30"},
env = {SENSOR_TYPE = "lidar"}
},
{
name = "example_node_py",
package = "example_package_py",
executable = "example_package_py.py",
language = "python",
args = {"--algorithm=detection"},
env = {PYTHONPATH = "${WORKSPACE}/lib:${WORKSPACE}/include"}
},
{
name = "example_node_java",
package = "example_ui_package_java",
executable = "ExampleDisplayAppJava",
language = "java",
args = {"--resolution=720p"},
env = {JAVA_OPTS = "-Xmx1g"}
},
{
name = "example_node_rust",
package = "example_package_rust",
executable = "example_package_rust",
language = "rust",
args = {"--mode=manual"},
env = {RUST_BACKTRACE = "1"}
},
{
name = "example_node_bridge_go",
package = "example_package_go",
executable = "example_package_go",
language = "go",
args = {"--port=9090"},
env = {GOMAXPROCS = "2"}
}
}
-- Define communication setup
topics = {
{
name = "/example/example",
type = "Example",
queue_size = 5,
publishers = {"example_node"},
subscribers = {"example1", "example2"}
}
}
-- Define services
services = {
{
name = "/example/example",
type = "Example",
server = "example_server",
clients = {"example_client"}
}
}
-- Define parameters
parameters = {
{
name = "/example/example",
type = "string",
value = global_config.system_name
}
}"""
try:
# Write the configuration to the file
with open(filename, 'w') as file:
file.write(config_content)
print(f"Successfully created example file {filename}")
return True
except Exception as e:
print(f"Error creating file: {e}")
return False
if __name__ == "__main__":
# Create the example Lua configuration file
success = create_example_lua_config_file()
if success:
print("Example configuration file creation completed.")
else:
print("Failed to create example configuration file.")

192
src/brsoa_generate_msgs.py Normal file
View File

@@ -0,0 +1,192 @@
# Copyright (C) 2025 Bellande Robotics Sensors 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/>.
import os
from bellande_parser.bellande_parser import Bellande_Format
def parse_def_file(filename):
bellande_parser = Bellande_Format()
content = bellande_parser.parse_bellande(filename)
structures = {}
enums = {}
current_struct = None
current_enum = None
for line in content.split('\n'):
line = line.strip()
if not line or line.startswith('#'):
continue
if line.startswith('struct'):
current_struct = line.split()[1]
structures[current_struct] = []
current_enum = None
elif line.startswith('enum'):
current_enum = line.split()[1]
enums[current_enum] = []
current_struct = None
elif current_struct:
field_type, field_name = line.split()
structures[current_struct].append((field_type, field_name))
elif current_enum:
enums[current_enum].append(line)
return structures, enums
def generate_cpp(structures, enums):
code = "#pragma once\n\n#include <string>\n#include <vector>\n#include <chrono>\n\nnamespace common_msgs {\n\n"
for enum_name, enum_values in enums.items():
code += f"enum class {enum_name} {{\n"
code += ',\n'.join(f" {value}" for value in enum_values)
code += "\n};\n\n"
for struct_name, fields in structures.items():
code += f"struct {struct_name} {{\n"
for field_type, field_name in fields:
if field_type == 'timestamp':
code += f" std::chrono::system_clock::time_point {field_name};\n"
elif field_type.startswith('list<'):
inner_type = field_type[5:-1]
code += f" std::vector<{inner_type}> {field_name};\n"
else:
code += f" {field_type} {field_name};\n"
code += "};\n\n"
code += "} // namespace common_msgs\n"
return code
def generate_python(structures, enums):
code = "from dataclasses import dataclass\nfrom typing import List\nimport datetime\n\n"
for enum_name, enum_values in enums.items():
code += f"class {enum_name}:\n"
for i, value in enumerate(enum_values):
code += f" {value} = {i}\n"
code += "\n"
for struct_name, fields in structures.items():
code += f"@dataclass\nclass {struct_name}:\n"
for field_type, field_name in fields:
if field_type == 'timestamp':
code += f" {field_name}: datetime.datetime\n"
elif field_type.startswith('list<'):
inner_type = field_type[5:-1]
code += f" {field_name}: List[{inner_type}]\n"
else:
code += f" {field_name}: {field_type}\n"
code += "\n"
return code
def generate_java(structures, enums):
code = "package common_msgs;\n\nimport java.time.Instant;\nimport java.util.List;\n\npublic class Messages {\n\n"
for enum_name, enum_values in enums.items():
code += f" public enum {enum_name} {{\n"
code += ',\n'.join(f" {value}" for value in enum_values)
code += "\n }\n\n"
for struct_name, fields in structures.items():
code += f" public static class {struct_name} {{\n"
for field_type, field_name in fields:
if field_type == 'timestamp':
code += f" public Instant {field_name};\n"
elif field_type.startswith('list<'):
inner_type = field_type[5:-1]
code += f" public List<{inner_type}> {field_name};\n"
else:
code += f" public {field_type} {field_name};\n"
code += " }\n\n"
code += "}\n"
return code
def generate_rust(structures, enums):
code = "use chrono::DateTime;\nuse chrono::Utc;\n\n"
for enum_name, enum_values in enums.items():
code += f"pub enum {enum_name} {{\n"
code += ',\n'.join(f" {value}" for value in enum_values)
code += "\n}\n\n"
for struct_name, fields in structures.items():
code += f"pub struct {struct_name} {{\n"
for field_type, field_name in fields:
if field_type == 'timestamp':
code += f" pub {field_name}: DateTime<Utc>,\n"
elif field_type.startswith('list<'):
inner_type = field_type[5:-1]
code += f" pub {field_name}: Vec<{inner_type}>,\n"
elif field_type == 'string':
code += f" pub {field_name}: String,\n"
else:
code += f" pub {field_name}: {field_type},\n"
code += "}\n\n"
return code
def generate_go(structures, enums):
code = "package common_msgs\n\nimport (\n\t\"time\"\n)\n\n"
for enum_name, enum_values in enums.items():
code += f"type {enum_name} int\n\nconst (\n"
for i, value in enumerate(enum_values):
code += f" {value} {enum_name} = iota\n"
code += ")\n\n"
for struct_name, fields in structures.items():
code += f"type {struct_name} struct {{\n"
for field_type, field_name in fields:
if field_type == 'timestamp':
code += f" {field_name.capitalize()} time.Time\n"
elif field_type.startswith('list<'):
inner_type = field_type[5:-1]
code += f" {field_name.capitalize()} []{inner_type}\n"
elif field_type == 'string':
code += f" {field_name.capitalize()} string\n"
else:
code += f" {field_name.capitalize()} {field_type}\n"
code += "}\n\n"
return code
def main():
bellande_parser = Bellande_Format()
config = bellande_parser.parse_bellande("project_config.bellande")
structures, enums = parse_def_file('common_msgs.bellande')
os.makedirs('build/common_msgs', exist_ok=True)
generators = {
'cpp': ('common_msgs.hpp', generate_cpp),
'python': ('common_msgs.py', generate_python),
'java': ('common_msgs.java', generate_java),
'rust': ('common_msgs.rs', generate_rust),
'go': ('common_msgs.go', generate_go),
}
for lang in config['languages']:
if lang in generators:
filename, generator = generators[lang]
with open(f'build/common_msgs/{filename}', 'w') as f:
f.write(generator(structures, enums))
else:
print(f"Warning: No generator for language '{lang}'")
if __name__ == "__main__":
main()