ENHANCEMENTS:

- MizEdit can call Python functions now.
- Carrier relocation script added (Credits to Mags and PyDCS)

BUGFIXES:
- .orig files will be written on /mission modify and /realweather also now

Author: karel26

core/data/impl/serverimpl.py

@@ -671,9 +671,6 @@ async def apply_mission_changes(self, filename: Optional[str] = None) -> str:
             dirty = False
             for ext in self.extensions.values():
                 if type(ext).beforeMissionLoad != Extension.beforeMissionLoad:
-                    # make an initial backup, if there is none
-                    if '.dcssb' not in filename and not os.path.exists(filename + '.orig'):
-                        shutil.copy2(filename, filename + '.orig')
                     new_filename, _dirty = await ext.beforeMissionLoad(new_filename)
                     if _dirty:
                         self.log.info(f'  => {ext.name} applied on {new_filename}.')
@@ -741,6 +738,9 @@ async def uploadMission(self, filename: str, url: str, force: bool = False, miss
     async def modifyMission(self, filename: str, preset: Union[list, dict]) -> str:
         miz = await asyncio.to_thread(MizFile, filename)
         await asyncio.to_thread(miz.apply_preset, preset)
+        # make an initial backup, if there is none
+        if '.dcssb' not in filename and not os.path.exists(filename + '.orig'):
+            shutil.copy2(filename, filename + '.orig')
         # write new mission
         new_filename = utils.create_writable_mission(filename)
         await asyncio.to_thread(miz.save, new_filename)

core/mizfile.py

@@ -1,4 +1,6 @@
 from __future__ import annotations
+
+import importlib
 import io
 import logging
 import luadata
@@ -461,6 +463,20 @@ def process_elements(reference: dict, **kwargs):
                         for _what in element.copy():
                             if utils.evaluate(config['delete'], **_what):
                                 element.remove(_what)
+                elif 'run' in config:
+                    if debug:
+                        self.log.debug(f"Processing {config['run']}() ...")
+                    module_name, func_name = config['run'].rsplit(".", 1)
+                    try:
+                        module = importlib.import_module(module_name)
+                        func = getattr(module, func_name)
+                        func(element, reference, **kwargs)
+                    except AttributeError:
+                        self.log.error(f"Function {func_name} not found in module {module_name}.")
+                        raise
+                    except ModuleNotFoundError:
+                        self.log.error(f"Module {module_name} not found.")
+                        raise
 
         def check_where(reference: dict, config: Union[list, str], debug: bool, **kwargs: dict) -> bool:
             if isinstance(config, str):
@@ -479,7 +495,11 @@ def check_where(reference: dict, config: Union[list, str], debug: bool, **kwargs
             for cfg in config:
                 self.modify(cfg)
             return
+
+        # enable debug logging
         debug = config.get('debug', False)
+
+        # which file has to be changed?
         file = config.get('file', 'mission')
         if file == 'mission':
             source = self.mission
@@ -490,13 +510,25 @@ def check_where(reference: dict, config: Union[list, str], debug: bool, **kwargs
         else:
             self.log.error(f"File {file} can not be changed.")
             return
+
         kwargs = {}
-        if 'variables' in config:
-            for name, value in config['variables'].items():
-                if value.startswith('$'):
-                    kwargs[name] = utils.evaluate(value, **kwargs)
-                else:
-                    kwargs[name] = next(utils.for_each(source, value.split('/'), debug=debug, **kwargs))
+        # check if we need to import stuff
+        for imp in config.get('imports', []):
+            try:
+                importlib.import_module(imp)
+            except ModuleNotFoundError:
+                self.log.error(f"Module '{imp}' could not be imported.")
+            except Exception as ex:
+                self.log.error(f"An error occurred while importing module '{imp}': {ex}")
+
+        # do we need to pre-set variables to work with?
+        for name, value in config.get('variables', {}).items():
+            if value.startswith('$'):
+                kwargs[name] = utils.evaluate(value, **kwargs)
+            else:
+                kwargs[name] = next(utils.for_each(source, value.split('/'), debug=debug, **kwargs))
+
+        # run the processing
         try:
             for_each = config['for-each'].lstrip('/')
         except KeyError:

core/utils/__init__.py

@@ -4,5 +4,6 @@
 from .dcs import *
 from .discord import *
 from .helper import *
+from .mizedit import *
 from .os import *
 from .performance import *

core/utils/helper.py

@@ -70,7 +70,8 @@
     "hash_password",
     "evaluate",
     "for_each",
-    "YAMLError"
+    "YAMLError",
+    "DictWrapper"
 ]
 
 logger = logging.getLogger(__name__)
@@ -923,3 +924,80 @@ class YAMLError(Exception):
     """
     def __init__(self, file: str, ex: Union[MarkedYAMLError, ValueError, SchemaError]):
         super().__init__(f"Error in {file}, " + ex.__str__().replace('"<unicode string>"', file))
+
+
+class DictWrapper:
+    """A wrapper for dictionaries enabling both attribute and key-based access."""
+
+    def __init__(self, data):
+        """Initialize with a dictionary or a list."""
+        if isinstance(data, dict):
+            self._data = {k: self._wrap(v) for k, v in data.items()}
+        elif isinstance(data, list):
+            self._data = [self._wrap(v) for v in data]
+        else:
+            self._data = data  # Handle non-dict types (e.g., primitive values)
+
+    @staticmethod
+    def _wrap(value):
+        """Wrap nested dictionaries or lists inside DictWrapper."""
+        if isinstance(value, dict):
+            return DictWrapper(value)
+        elif isinstance(value, list):
+            return [DictWrapper._wrap(v) for v in value]
+        return value
+
+    def __getattr__(self, name):
+        """Access dictionary keys as attributes."""
+        try:
+            return self._data[name]
+        except KeyError:
+            raise AttributeError(f"Attribute '{name}' not found")
+
+    def __setattr__(self, name, value):
+        """Set dictionary keys as attributes."""
+        if name == "_data":
+            super().__setattr__(name, value)
+        else:
+            self._data[name] = self._wrap(value)
+
+    def __delattr__(self, name):
+        """Delete keys like attributes."""
+        try:
+            del self._data[name]
+        except KeyError:
+            raise AttributeError(f"Attribute '{name}' not found")
+
+    def __getitem__(self, key):
+        """Support list-style or dictionary-style key/item access."""
+        return self._data[key]
+
+    def __setitem__(self, key, value):
+        """Set items using key."""
+        self._data[key] = self._wrap(value)
+
+    def __delitem__(self, key):
+        """Support item deletion."""
+        del self._data[key]
+
+    def __iter__(self):
+        """Allow iteration."""
+        return iter(self._data)
+
+    def __repr__(self):
+        """Pretty representation."""
+        return repr(self._data)
+
+    def to_dict(self):
+        def _unwrap_list(value):
+            if isinstance(value, list):
+                return [(v.to_dict() if isinstance(v, DictWrapper) else _unwrap_list(v)) for v in value]
+            return value
+
+        if isinstance(self._data, dict):
+            return {
+                k: (v.to_dict() if isinstance(v, DictWrapper) else _unwrap_list(v)) for k, v in self._data.items()
+            }
+        elif isinstance(self._data, list):
+            return _unwrap_list(self._data)
+        return self._data

core/utils/mizedit/__init__.py

@@ -0,0 +1 @@
+from .carrier_relocator import *

core/utils/mizedit/carrier_relocator.py

@@ -0,0 +1,159 @@
+# Credits to magwo (Magnus Wolffelt)
+import math
+
+from core.utils.helper import DictWrapper
+from .trigonometric_carrier_cruise import get_ship_course_and_speed
+from .me_utils import Speed, Distance, Heading, knots, HeadingAndSpeed
+
+__all__ = ['relocate_carrier']
+
+
+#
+# TAKEN FROM PyDCS! >>> START
+#
+def point_from_heading(_x: float, _y: float, heading: float, distance: float) -> tuple[float, float]:
+    """Calculates a point from a given point, heading and distance.
+
+    :param _x: source point x
+    :param _y: source point y
+    :param heading: heading in degrees from source point
+    :param distance: distance from source point
+    :return: returns a tuple (x, y) of the calculated point
+    """
+    while heading < 0:
+        heading += 360
+    heading %= 360
+    rad_heading = math.radians(heading)
+    x = _x + math.cos(rad_heading) * distance
+    y = _y + math.sin(rad_heading) * distance
+
+    return x, y
+
+
+def heading_between_points(x1: float, y1: float, x2: float, y2: float) -> float:
+    """Returns the angle between 2 points in degrees.
+
+    :param x1: x coordinate of point 1
+    :param y1: y coordinate of point 1
+    :param x2: x coordinate of point 2
+    :param y2: y coordinate of point 2
+    :return: angle in degrees
+    """
+    def angle_trunc(a):
+        while a < 0.0:
+            a += math.pi * 2
+        return a
+    deltax = x2 - x1
+    deltay = y2 - y1
+    return math.degrees(angle_trunc(math.atan2(deltay, deltax)))
+
+
+def distance_to_point(x1: float, y1: float, x2: float, y2: float) -> float:
+    """Returns the distance between 2 points
+
+    :param x1: x coordinate of point 1
+    :param y1: y coordinate of point 1
+    :param x2: x coordinate of point 2
+    :param y2: y coordinate of point 2
+    :return: distance in point units(m)
+    """
+    return math.hypot(x2 - x1, y2 - y1)
+#
+# TAKEN FROM PyDCS <<<< END
+#
+
+def get_carrier_cruise(wind: dict, deck_angle: float, desired_apparent_wind: Speed) -> HeadingAndSpeed:
+    wind_speed = Speed.from_meters_per_second(wind.get('speed', 0))
+    heading, speed, apparent_wind_angle = get_ship_course_and_speed(
+        wind.get('dir', 0), wind_speed.knots, desired_apparent_wind.knots
+    )
+    # Quick hack for Tarawa
+    if deck_angle == 0:
+        wind_heading = Heading( wind.get('dir', 0))
+        heading = wind_heading.opposite.degrees
+
+    solution = HeadingAndSpeed(
+        Heading.from_degrees(heading), Speed.from_knots(speed)
+    )
+    return solution
+
+
+def rotate_group_around(group: DictWrapper, pivot: tuple[float, float], degrees_change: float):
+    # My fear was that using sin/cos would result in incorrect
+    # transforms when not near the equator. Polar coordinates (heading + distance)
+    # should resolve that, if the Point functions are implemented correctly.
+    # I think they're not, but this code at least doesn't prevent correct transform.
+    # Maybe DCS doesn't even use mercator projection.
+    for unit in group.units:
+        distance = distance_to_point(pivot[0], pivot[1], unit.x, unit.y)
+        heading = heading_between_points(pivot[0], pivot[1], unit.x, unit.y)
+        new_heading = Heading.from_degrees(heading + degrees_change).degrees
+
+        unit.x, unit.y = point_from_heading(pivot[0], pivot[1], new_heading, distance)
+        unit.heading = Heading.from_degrees(unit.heading + degrees_change).degrees
+
+
+def relocate_carrier(_: dict, reference: dict, **kwargs):
+    # create a wrapper, to make it easier (and to mainly keep the old code)
+    group = DictWrapper(reference)
+    route = group.route
+
+    wind = kwargs.get('wind', {})
+    carrier = group.units[0]
+    deck_angle = 0 if carrier.type == 'LHA_Tarawa' else -9.12
+    cruise = get_carrier_cruise(wind, deck_angle, Speed.from_knots(25))
+
+    radius = Distance.from_nautical_miles(50)
+    carrier_start_pos = point_from_heading(
+        group['x'], group['y'], cruise.heading.opposite.degrees, radius.meters
+    )
+    carrier_end_pos = point_from_heading(reference['x'], reference['y'], cruise.heading.degrees, radius.meters)
+
+    group_heading_before_change = heading_between_points(
+        route.points[0].x, route.points[0].y, route.points[1].x, route.points[1].y
+    )
+    group_position_before_change = (route.points[0].x, route.points[0].y)
+
+    if len(route.points) < 4:
+        print(f"Carrier group {reference['name']} missing waypoint")
+        return
+        # TODO
+
+    route.points[0].x = carrier_start_pos[0]
+    route.points[0].y = carrier_start_pos[1]
+    route.points[0].speed = cruise.speed.meters_per_second
+
+    route.points[1].x = carrier_end_pos[0]
+    route.points[1].y = carrier_end_pos[1]
+    route.points[1].ETA_locked = False
+    route.points[1].speed = cruise.speed.meters_per_second
+    route.points[1].speed_locked = True
+
+    route.points[2].x = carrier_start_pos[0]
+    route.points[2].y = carrier_start_pos[1]
+    route.points[2].speed = knots(50).meters_per_second
+    route.points[2].ETA_locked = False
+    route.points[2].speed_locked = True
+
+    route.points[3].x = carrier_end_pos[0]
+    route.points[3].y = carrier_end_pos[1]
+    route.points[3].ETA_locked = False
+    route.points[3].speed = cruise.speed.meters_per_second
+    route.points[3].speed_locked = True
+
+    position_change = (
+        carrier_start_pos[0] - group_position_before_change[0],
+        carrier_start_pos[1] - group_position_before_change[1]
+    )
+    for unit in group.units:
+        unit.x = unit.x + position_change[0]
+        unit.y = unit.y + position_change[1]
+
+    heading_change = cruise.heading.degrees - group_heading_before_change
+    rotate_group_around(
+        group, (route.points[0].x, route.points[0].y), heading_change
+    )
+
+    # change the real thing
+    reference['route'] = route.to_dict()
+    reference['units'] = [unit.to_dict() for unit in group.units]