from __future__ import annotations
from dataclasses import dataclass, field, asdict
from types import NoneType
from typing import Any, Callable, cast, Protocol, Sequence, TypeAlias, TypeGuard, TYPE_CHECKING
from threading import RLock as threading_RLock
from functools import wraps as functools_wraps

import enum, subprocess, re, os, sys

def locked(method):
    @functools_wraps(method)
    def wrapper(self, *args, **kwargs):
        with self._lock:
            return method(self, *args, **kwargs)
    return wrapper

dll_dir_handles = []

if TYPE_CHECKING:
    import pandas as pd
    import multiprocessing.connection

class Ga3Error(Exception):
    pass

class ArrayLike(Protocol):
    """Numpy-compatible array object passed between GA3 and Python code."""
    ndim: int
    shape: tuple[int, ...]
    def __array__(self, dtype=None): ...

CellData: TypeAlias = int|float|str|None
ColData: TypeAlias = list[int|None]|list[float|None]|list[str|None]|ArrayLike
MultiColData: TypeAlias = ColData|list[ColData]|ArrayLike
SafeColSpec: TypeAlias = list[str]|list[int]|re.Pattern|Callable[[str, dict[str, Any]], bool]|None

class LimTableLike(Protocol):
    """GA3 table object used to exchange tabular data and table schemas."""
    tableName: str
    tableMetadata: dict[str, Any]
    colIdList: list[str]
    colMetadataList: list[dict[str, Any]]
    def safeColIndexList(self, cols: SafeColSpec = None, *, not_found: list[str|int]|None = None) -> list[int]: ...
    def colIndexById(self, colId: str) -> int: ...
    def colArray(self, cols: str|int|list[str]|list[int]|None = None) -> ArrayLike|list[ArrayLike]: ...
    def colData(self, cols: list[str]|list[int]|None = None) -> list[int|None]|list[float|None]|list[str|None]: ...
    def setTableName(self, name : str) -> None: ...
    def setTableMetadata(self, meta : dict) -> None: ...
    def setColData(self, cols: SafeColSpec, data: MultiColData) -> None: ...
    def removeCol(self, at: int|str|None = None) -> None: ...
    def removeAllCols(self) -> None: ...
    def insertCol(self, id: str, data: list, meta: dict|None = None, *, before: int|str|None = None, after: int|str|None = None) -> None: ...
    def pandasDataFrame(self, cols: SafeColSpec = None, *, use_col_ids: bool = False) -> "pd.DataFrame": ...
    def pandasMetaDataFrame(self) -> "pd.DataFrame": ...
    def setPandasDataFrame(self, df: "pd.DataFrame", cols: SafeColSpec = None, *, not_found: list[str]|None = None) -> list[str]: ...
    def insertPrivateTable(self, table : LimTableLike) -> None: ...

class MplFigureLike(Protocol):
    """Matplotlib-like figure or canvas that can be stored as a GA3 result image."""
    def print_png(*args, **kwargs) -> None: ...
    def savefig(*args, **kwargs) -> None: ...

AnyColorDef: TypeAlias = int|str|tuple[int, int, int]

def _is_str_list(x: object) -> TypeGuard[list[str]]:
    return isinstance(x, list) and all(isinstance(i, str) for i in x)

def _is_int_list(x: object) -> TypeGuard[list[str]]:
    return isinstance(x, list) and all(isinstance(i, str) for i in x)

def _is_list_list(x: object) -> TypeGuard[list[list]]:
    return isinstance(x, list) and all(isinstance(i, list) for i in x)

def _is_1d_array_list(x: object) -> TypeGuard[list[ArrayLike]]:
    return isinstance(x, list) and all((getattr(i, '__array__', False) and i.ndim == 1) for i in x)

def _is_1d_array(x: object) -> TypeGuard[ArrayLike]:
    return getattr(x, '__array__', False) and cast(ArrayLike, x).ndim == 1

def _is_2d_array(x: object) -> TypeGuard[ArrayLike]:
    return getattr(x, '__array__', False) and cast(ArrayLike, x).ndim == 2

@dataclass(frozen=True, kw_only=True)
class Node:
    """GA3 processing node that owns the parameter represented by this object."""
    name: str
    uuid: str
    defaultName: str

class ExperimentLoopType(enum.IntEnum):
    """GA3 experiment loop kind, such as well, site, time, or Z-stack."""
    loopTypeUnknown  = 0
    loopTypePlate    = 1
    loopTypeWell     = 2
    loopTypeXYSite   = 3
    loopTypeTLapse   = 4
    loopTypeZStack   = 5
    loopTypeCount    = 6
    loopTypeSlide    = 7
    loopTypeRegion   = 8
    loopTypeTemporary= 9

@dataclass(frozen=True, kw_only=True)
class LoopDef:
    """One GA3 experiment loop available to a Python node invocation."""
    name: str
    type: ExperimentLoopType
    index: int

    def __post_init__(self):
        if type(self.type) == int:
            object.__setattr__(self, 'type', ExperimentLoopType(self.type))

@dataclass(frozen=True, kw_only=True)
class ParameterDef:
    """Common GA3 parameter definition metadata for input and output ports."""
    parentNode: Node
    parName: str
    parUuid: str
    parIsInput: bool
    loopDefList: list[LoopDef]

    def __post_init__(self):
        if type(self.parentNode) == dict:
            object.__setattr__(self, 'parentNode', Node(**self.parentNode))
        if type(self.loopDefList) == list and all(type(item) == dict for item in self.loopDefList):
            object.__setattr__(self, 'loopDefList', [LoopDef(**item) for item in cast(dict, self.loopDefList)])

    def loopNames(self, loop_type: ExperimentLoopType|list[ExperimentLoopType]|None = None) -> list[str]:
        """Return loop names available to this parameter.

        Args:
            loop_type: Optional loop type, list of loop types, or None for all loops.

        Returns:
            List of loop names, preserving the GA3 loop order.

        Raises:
            TypeError: if loop_type is not None, ExperimentLoopType, or a list of ExperimentLoopType.
        """
        if loop_type is None:
            return [ loop.name for loop in self.loopDefList]
        elif isinstance(loop_type, ExperimentLoopType):
            return [ loop.name for loop in self.loopDefList if loop.type == loop_type]
        elif isinstance(loop_type, list) and all(isinstance(lt, ExperimentLoopType) for lt in loop_type):
            return [ loop.name for loop in self.loopDefList if loop.type in loop_type]
        else:
            raise TypeError(f"Expected ExperimentLoopType|list[ExperimentLoopType]|None, got {type(loop_type).__name__}")

    def format_coordinates(self, coordinates: tuple[int], *, sep: str = '_', one_based_indexes: bool = True, leading_zeros: int = 0) -> str:
        """Format loop coordinates into a compact label.

        Use this when naming files, result rows, or diagnostic messages from
        ctx.outCoordinates or ctx.inpParameterCoordinates.

        Args:
            coordinates: Zero-based coordinate tuple in the same order as loopDefList.
            sep: Separator inserted between coordinate parts.
            one_based_indexes: Convert indexes to one-based labels when True.
            leading_zeros: Minimum number of digits for each index.

        Returns:
            Coordinate label such as "W01_T03_Z02".
        """
        name = [ '?', 'P', 'W', 'M', 'T', 'Z', 'C', 'S', 'R', 'X' ]
        return sep.join(f'{name[loop.type]}{coord+(1 if one_based_indexes else 0):0{leading_zeros}}' for loop, coord in zip(self.loopDefList, coordinates))

@dataclass(frozen=True, kw_only=True)
class Metadata:
    """Image metadata shared by GA3 channel and binary parameters."""
    componentCount: int
    bitsPerComponent: int
    size: tuple[int, int, int]
    calibration: tuple[float, float, float]
    alignment: int
    isRGB: bool

    @property
    def width(self) -> int:
        return self.size[0]

    @property
    def height(self) -> int:
        return self.size[1]

    @property
    def depth(self) -> int:
        return self.size[2]

    @property
    def calibrationXY(self) -> float:
        return self.calibration[0]

    @property
    def calibrationZ(self) -> float:
        return self.calibration[2]

    @property
    def calibrated(self) -> tuple[bool, ...]:
        return (0 < self.calibration[0], 0 < self.calibration[1], 0 < self.calibration[2])
    @property
    def units(self) -> tuple[str, ...]:
        return tuple("\xB5m" if cal else "px" for cal in self.calibrated)

    def withDifferentSize(self, size: tuple[int, int, int], calibration: tuple[float, float, float]|None = None) -> Metadata:
        """Set output metadata to a different image size.

        Call this in output(...) when an output image or binary has dimensions
        that differ from the input used to initialize the output definition.

        Args:
            size: Output size as (X, Y, Z).
            calibration: Optional calibration tuple as (X, Y, Z).

        Returns:
            self
        """
        assert isinstance(size, tuple) and isinstance(calibration, (tuple, NoneType)), "Unexpected argument type"
        assert len(size) == 3, "Unexpected length of argument 'size'"
        object.__setattr__(self, 'size', size)
        if calibration is not None:
            assert len(calibration) == 3, "Unexpected length of argument 'calibration'"
            if type(calibration) == tuple and len(calibration) == 3:
                object.__setattr__(self, 'calibration', calibration)
            else:
                raise TypeError("Expected calibration: tuple[int, int, int]")
        return self

    def copyMetadataFrom(self, other: Metadata) -> Metadata:
        assert isinstance(other, Metadata), "Unexpected argument type"
        object.__setattr__(self, 'componentCount', other.componentCount)
        object.__setattr__(self, 'bitsPerComponent', other.bitsPerComponent)
        object.__setattr__(self, 'size', other.size)
        object.__setattr__(self, 'calibration', other.calibration)
        object.__setattr__(self, 'alignment', other.alignment)
        object.__setattr__(self, 'isRGB', other.isRGB)
        return self

@dataclass(frozen=True, kw_only=True)
class InputChannelDef(ParameterDef, Metadata):
    """Read-only definition of a GA3 input channel available in output(...).

    Use this to inspect the input channel name, color, size, calibration, and
    bit depth when defining outputs. Do not modify input parameter definitions.
    """
    channelName: str
    channelColor: int

@dataclass(frozen=True, kw_only=True)
class OutputChannelDef(ParameterDef, Metadata):
    """Writable definition of a GA3 output channel configured in output(...)."""
    assignedParName: str|None = None
    channelName: str|None = None
    channelColor: int|None = None

    def assign(self, param: InputChannelDef) -> OutputChannelDef:
        """Make this output channel reuse an input channel definition.

        Use this when the output should keep the input channel name, color,
        size, calibration, component count, and bit depth.

        Args:
            param: Input channel definition to assign.

        Returns:
            self
        """
        assert isinstance(param, InputChannelDef), "Unexpected argument type"
        object.__setattr__(self, 'assignedParName', param.parName)
        object.__setattr__(self, 'channelName', param.channelName)
        object.__setattr__(self, 'channelColor', param.channelColor)
        return self

    def makeNewMono(self, name: str, color: AnyColorDef) -> OutputChannelDef:
        """Define a new mono output channel.

        Args:
            name: Channel name shown in GA3.
            color: Channel color as "#RRGGBB", integer RGB, or (R, G, B).

        Returns:
            self
        """
        assert isinstance(name, str) and isinstance(color, (int, str, tuple)), "Unexpected argument type"
        object.__setattr__(self, 'assignedParName', None)
        object.__setattr__(self, 'channelName', name)
        object.__setattr__(self, 'channelColor', _parse_color(color))
        object.__setattr__(self, 'componentCount', 1)
        object.__setattr__(self, 'isRGB', False)
        return self

    def makeNewRgb(self, name: str|None = None) -> OutputChannelDef:
        """Define a new RGB output channel.

        Args:
            name: Channel name shown in GA3.

        Returns:
            self
        """
        assert isinstance(name, str), "Unexpected argument type"
        object.__setattr__(self, 'assignedParName', None)
        object.__setattr__(self, 'channelName', name)
        object.__setattr__(self, 'channelColor', None)
        object.__setattr__(self, 'componentCount', 3)
        object.__setattr__(self, 'isRGB', True)
        return self

    def makeFloat(self) -> OutputChannelDef:
        object.__setattr__(self, 'bitsPerComponent', 32)
        return self

    def makeUInt16(self) -> OutputChannelDef:
        object.__setattr__(self, 'bitsPerComponent', 16)
        return self

    def makeUInt8(self) -> OutputChannelDef:
        object.__setattr__(self, 'bitsPerComponent', 8)
        return self

    def copyFrom(self, other: OutputChannelDef) -> OutputChannelDef:
        assert isinstance(other, OutputChannelDef), "Unexpected argument type"
        object.__setattr__(self, 'assignedParName', other.assignedParName)
        object.__setattr__(self, 'channelName', other.channelName)
        object.__setattr__(self, 'channelColor', other.channelColor)
        self.copyMetadataFrom(other)
        return self

@dataclass(frozen=True, kw_only=True)
class InputBinaryDef(ParameterDef, Metadata):
    """Read-only definition of a GA3 input binary available in output(...).

    Use this to inspect the input binary name, color, size, calibration, and
    bit depth when defining outputs. Do not modify input parameter definitions.
    """
    binaryName: str
    binaryColor: int

@dataclass(frozen=True, kw_only=True)
class OutputBinaryDef(ParameterDef, Metadata):
    """Writable definition of a GA3 output binary configured in output(...)."""
    assignedParName: str|None = None
    binaryName: str|None = None
    binaryColor: int|None = None

    def assign(self, param: InputBinaryDef) -> OutputBinaryDef:
        """Make this output binary reuse an input binary definition.

        Use this when the output should keep the input binary name, color,
        size, calibration, component count, and bit depth.

        Args:
            param: Input binary definition to assign.

        Returns:
            self
        """
        assert isinstance(param, InputBinaryDef), "Unexpected argument type"
        object.__setattr__(self, 'assignedParName', param.parName)
        object.__setattr__(self, 'binaryName', param.binaryName)
        object.__setattr__(self, 'binaryColor', param.binaryColor)
        return self

    def makeNew(self, name: str, color: AnyColorDef) -> OutputBinaryDef:
        """Define a new binary output.

        Args:
            name: Binary name shown in GA3.
            color: Binary color as "#RRGGBB", integer RGB, or (R, G, B).

        Returns:
            self
        """
        assert isinstance(name, str) and isinstance(color, (int, str, tuple)), "Unexpected argument type"
        object.__setattr__(self, 'assignedParName', None)
        object.__setattr__(self, 'binaryName', name)
        object.__setattr__(self, 'binaryColor', _parse_color(color))
        object.__setattr__(self, 'componentCount', 1)
        object.__setattr__(self, 'isRGB', False)
        return self

    def makeInt32(self) -> OutputBinaryDef:
        object.__setattr__(self, 'bitsPerComponent', 32)
        return self

    def makeUInt8(self) -> OutputBinaryDef:
        object.__setattr__(self, 'bitsPerComponent', 8)
        return self

    def copyFrom(self, other: OutputBinaryDef) -> OutputBinaryDef:
        assert isinstance(other, OutputBinaryDef), "Unexpected argument type"
        object.__setattr__(self, 'assignedParName', other.assignedParName)
        object.__setattr__(self, 'binaryName', other.binaryName)
        object.__setattr__(self, 'binaryColor', other.binaryColor)
        self.copyMetadataFrom(other)
        return self

@dataclass(frozen=True, kw_only=True)
class InputTableDef(ParameterDef):
    """Read-only definition of a GA3 input table available in output(...).

    Use this to inspect or copy the input table schema and accumulation
    metadata when defining outputs. Do not modify input parameter definitions.
    """
    tableDef: LimTableLike

    def accumulatedOver(self) -> list[str]:
        """Return names of loops already accumulated in this input table."""
        return self.tableDef.tableMetadata.get('accumulatedLoops', [])

@dataclass(frozen=True, kw_only=True)
class OutputTableDef(ParameterDef):
    """Writable definition of a GA3 output table configured in output(...)."""
    assignedParName: str
    tableDef: LimTableLike

    def assign(self, inParam: InputTableDef) -> OutputTableDef:
        """Make this output table reuse an input table schema.

        Args:
            inParam: Input table definition to assign.

        Returns:
            self
        """
        assert isinstance(inParam, InputTableDef), "inParam MUST be a Table (InputTableDef)"
        object.__setattr__(self, 'assignedParName', inParam.parName)
        object.__setattr__(self, 'tableDef', inParam.tableDef)
        return self

    def makeEmpty(self, name: str, inParam: InputTableDef|None = None) -> OutputTableDef:
        """Create a named output table with no copied columns.

        If inParam is an accumulated input table, the accumulation metadata is
        preserved so GA3 interprets the output rows at the same loop level.

        Args:
            name: Table name shown in GA3.
            inParam: Optional input table whose accumulation metadata should be preserved.

        Returns:
            self
        """
        assert isinstance(name, str), "name must be a string (str)"
        assert (inParam is None) or isinstance(inParam, InputTableDef), "inParam MUST be a Table or None (InputTableDef|None)"
        object.__setattr__(self, 'assignedParName', None)
        self.tableDef.setTableName(name)
        self.tableDef.removeAllCols()
        if inParam is not None and inParam.tableDef.tableMetadata.get('contains') == 'accumulatedFrames':
            self.tableDef.tableMetadata['contains'] = 'accumulatedFrames'
            self.tableDef.tableMetadata['accumulatedLoops'] = inParam.tableDef.tableMetadata['accumulatedLoops']
        return self

    def makeNew(self, name: str, inParam: InputTableDef|None = None) -> OutputTableDef:
        """Create a named output table.

        With an input table, the input schema is copied and renamed. Without an
        input table, loop-index columns are created automatically.

        Args:
            name: Table name shown in GA3.
            inParam: Optional input table definition to copy.

        Returns:
            self
        """
        assert isinstance(name, str), "name must be a string (str)"
        assert (inParam is None) or isinstance(inParam, InputTableDef), "inParam MUST be a Table or None (InputTableDef|None)"
        self.makeEmpty(name, inParam)
        if inParam is not None:
            if isinstance(inParam, InputTableDef):
                tableDef = inParam.tableDef
                tableDef.setTableName(name)
                object.__setattr__(self, 'tableDef', tableDef)
            else:
                raise TypeError('Expected param: str|InputTableDef')
        else:
            self.withLoopCols()
        return self

    def makeResult(self, name: str, inParam: InputTableDef|None = None) -> OutputTableDef:
        """Create a result-table shell for rendered outputs.

        Use this before OutputTableData.withMplImage(...) or other result
        helpers that write HTML and InitialState rows.

        Args:
            name: Result table name shown in GA3.
            inParam: Optional input table definition to copy.

        Returns:
            self
        """
        self.makeNew(name, inParam)
        self.tableDef.tableMetadata['_systemFlags'] = [ 'result', 'webpage', 'html', 'javaScript', 'css', 'jsonInitialState' ]
        self._withColumn("QString", "Html", None, "htmlCode", "htmlCode", dict(url="qrc:///gnr_express/WebResults/v02/results.html", ver=2))
        self._withColumn("QString", "InitialState", None, "jsonInitialState", "jsonInitialState", dict(ver=2))
        return self

    def withLoopCols(self, *, newColNames: list[str]|None = None) -> OutputTableDef:
        """Add GA3 loop-index columns to the output table.

        Args:
            newColNames: Optional list that receives added column titles.

        Returns:
            self
        """
        for loop in self.loopDefList:
            id: str = f"_loop{loop.name}Index"
            name: str = f"{loop.name}Index"
            assert id not in self.tableDef.colIdList, "Loop columns already defined perhaps from makeNew(...)"
            self._withColumn("int", name, id=id, meta=dict(loopName=loop.name, loopType=int(loop.type)))
            if isinstance(newColNames, list):
                newColNames.append(name)
        return self

    def withEntityCol(self, *, newColNames: list[str]|None = None) -> OutputTableDef:
        """Add the GA3 Entity column used by object-linked table rows."""
        self._withColumn("QString", "Entity", id="_Entity")
        if isinstance(newColNames, list):
            newColNames += [ "Entity" ]
        return self

    def withObjectCols(self, *, newColNames: list[str]|None = None) -> OutputTableDef:
        """Add 2D object-link columns Entity and ObjectId."""
        self._withColumn("QString", "Entity", id="_Entity")
        self._withColumn("int", "ObjectId", id="_ObjId")
        if isinstance(newColNames, list):
            newColNames += [ "Entity", "ObjectId" ]
        return self

    def withObject3dCols(self, *, newColNames: list[str]|None = None) -> OutputTableDef:
        """Add 3D object-link columns Entity and ObjectId."""
        self._withColumn("QString", "Entity", id="_Entity")
        self._withColumn("int", "ObjectId", id="_ObjId3d")
        if isinstance(newColNames, list):
            newColNames += [ "Entity", "ObjectId" ]
        return self

    def withIntCol(self, title: str, unit: str|None = None, feature:str|None = None, id: str|None = None) -> OutputTableDef:
        """Add an integer column to the output table."""
        assert isinstance(title, str) and isinstance(unit, (str, NoneType)) and isinstance(feature, (str, NoneType)) and isinstance(id, (str, NoneType)), "Unexpected argument type"
        return self._withColumn("int", title, unit, feature, id)

    def withFloatCol(self, title: str, unit: str|None = None, feature:str|None = None, id: str|None = None) -> OutputTableDef:
        """Add a floating-point column to the output table."""
        assert isinstance(title, str) and isinstance(unit, (str, NoneType)) and isinstance(feature, (str, NoneType)) and isinstance(id, (str, NoneType)), "Unexpected argument type"
        return self._withColumn("double", title, unit, feature, id)

    def withStringCol(self, title: str, unit: str|None = None, feature:str|None = None, id: str|None = None) -> OutputTableDef:
        """Add a string column to the output table."""
        assert isinstance(title, str) and isinstance(unit, (str, NoneType)) and isinstance(feature, (str, NoneType)) and isinstance(id, (str, NoneType)), "Unexpected argument type"
        return self._withColumn("QString", title, unit, feature, id)

    def accumulatedOver(self) -> list[str]:
        """Return names of loops this output table is marked as accumulated over."""
        return self.tableDef.tableMetadata.get('accumulatedLoops', [])

    def setAccumulatedOver(self, *args) -> OutputTableDef:
        """Mark this table as accumulated over the named loops.

        Args:
            *args: Loop names to accumulate over.

        Returns:
            self
        """
        assert 0 < len(args) and all(isinstance(a, str) for a in args), f"All arguments must be loop names (tuple[str])"
        self.tableDef.tableMetadata['contains'] = 'accumulatedFrames'
        self.tableDef.tableMetadata['accumulatedLoops'] = list(set(list(args) + self.accumulatedOver()))
        return self

    def setUnaccumulated(self) -> OutputTableDef:
        """Mark this table as a per-frame table."""
        assert 'contains' not in self.tableDef.tableMetadata or self.tableDef.tableMetadata['contains'] == 'frame', "Table is already accumulated"
        self.tableDef.tableMetadata['contains'] = 'frame'
        del self.tableDef.tableMetadata['accumulatedLoops']
        return self

    def setAccumulatedOverZStack(self) -> OutputTableDef:
        """Mark this table as accumulated over Z-stack loops."""
        loop_names = self.loopNames(ExperimentLoopType.loopTypeZStack)
        self.setAccumulatedOver(*loop_names)
        return self

    def setAccumulatedOverTimeLapse(self) -> OutputTableDef:
        """Mark this table as accumulated over time-lapse loops."""
        loop_names = self.loopNames(ExperimentLoopType.loopTypeTLapse)
        self.setAccumulatedOver(*loop_names)
        return self

    def setAccumulatedOverMultiPoint(self) -> OutputTableDef:
        """Mark this table as accumulated over multipoint/site loops."""
        loop_names = self.loopNames(ExperimentLoopType.loopTypeXYSite)
        self.setAccumulatedOver(*loop_names)
        return self

    def setAccumulatedOverAll(self) -> OutputTableDef:
        """Mark this table as accumulated over every available loop."""
        loop_names = self.loopNames(None)
        self.setAccumulatedOver(*loop_names)
        return self

    def _withColumn(self, decltype: str, title: str, unit: str|None = None, feature:str|None = None, id: str|None = None, meta: dict[str, Any]|None = None) -> OutputTableDef:
        assert isinstance(decltype, str) and isinstance(title, str) and isinstance(unit, (str, NoneType)) and isinstance(feature, (str, NoneType)) and isinstance(id, (str, NoneType)) and isinstance(meta, (dict, NoneType)), "Unexpected argument type"
        if id is None:
            id = f'{self.parentNode.uuid}.{self.parentNode.defaultName}.{feature}' if feature is not None else f'{self.parentNode.uuid}.{self.parentNode.defaultName}.{title}'
        if meta is None:
            meta = dict()
        meta.update(dict(decltype=decltype, title=title))
        if unit is not None:
            meta["units"] = unit
        if feature is not None:
            meta["feature"] = feature
        self.tableDef.insertCol(id, [], meta)
        return self

    def copyFrom(self, other: OutputTableDef) -> OutputTableDef:
        assert isinstance(other, OutputTableDef), "Unexpected argument type"
        object.__setattr__(self, 'assignedParName', other.assignedParName)
        object.__setattr__(self, 'tableDef', other.tableDef)
        return self

## Param data classes

@dataclass(frozen=True, kw_only=True)
class ArrayData:
    """Runtime GA3 image array container used by channel and binary data ports."""
    data: ArrayLike|None

    # internal method to set an array in shared memory for out-proc
    # for setting data use data[:]
    def _setData(self, new_data: ArrayLike|None) -> None:
        assert new_data is None or getattr(new_data, '__array__', False),  "Unexpected argument type"
        object.__setattr__(self, 'data', new_data)

@dataclass(frozen=True, kw_only=True)
class RecordedData:
    """Runtime GA3 recorded-data side table attached to channel and binary data."""
    recdata: LimTableLike|None

    def _copyRecordedDataFrom(self, other: RecordedData) -> None:
        assert isinstance(other, RecordedData), "Unexpected argument type"
        if other.recdata is not None and self.recdata is not None:
            col_ids = other.recdata.colIdList
            col_data = other.recdata.colData(col_ids)
            self.recdata.setColData(col_ids, col_data)


@dataclass(frozen=True, kw_only=True)
class InputChannelData(ParameterDef, Metadata, ArrayData, RecordedData):
    """Read-only GA3 input channel data available in run(...).

    Read the data array and metadata, but do not replace or modify the input
    array. Write results through the corresponding output data object instead.
    """
    channelName: str

@dataclass(frozen=True, kw_only=True)
class OutputChannelData(ParameterDef, Metadata, ArrayData, RecordedData):
    """Writable GA3 output channel data filled in run(...)."""
    def copyRecordedDataFrom(self, other: RecordedData) -> None:
        """Copy recorded per-pixel/per-object data from another channel or binary."""
        self._copyRecordedDataFrom(other)


@dataclass(frozen=True, kw_only=True)
class InputBinaryData(ParameterDef, Metadata, ArrayData, RecordedData):
    """Read-only GA3 input binary data available in run(...).

    Read the data array and metadata, but do not replace or modify the input
    array. Write results through the corresponding output data object instead.
    """
    binaryName: str

@dataclass(frozen=True, kw_only=True)
class OutputBinaryData(ParameterDef, Metadata, ArrayData, RecordedData):
    """Writable GA3 output binary data filled in run(...)."""
    def copyRecordedDataFrom(self, other: RecordedData) -> None:
        """Copy recorded per-pixel/per-object data from another channel or binary."""
        self._copyRecordedDataFrom(other)

@dataclass(frozen=True, kw_only=True)
class InputTableData(ParameterDef, RecordedData):
    """Read-only GA3 input table data available in run(...).

    Use helper methods to read columns or DataFrames from the table. Do not
    modify input table data; write derived rows through OutputTableData.
    """
    data: LimTableLike

    def colArray(self, cols: str|int|SafeColSpec = None) -> ArrayLike|list[ArrayLike]:
        """Return numpy ndarray or a list there of for specified columns.

        If cols is scalar ndarray is returned otherwise a list is returned instead.

        Column specifier:
        - column name, ID or index returns single column
        - list of column names, IDs or indexes returns list of columns (must be present)
        - None return all columns
        - re.Pattern return columns where fullmatch succeeded on column ID or Title
        - Callable[[str, dict], bool] returns columns where callable returns True for (ID, Metadata)

        When column names, IDs or indexes are specified explicitly in a list they must
        be present otherwise and AssertError is raised.

        Args:
            cols: col name, id or index or a list there of or a Pattern or Callable

        Returns:
            Single 1D numpy ndarray or a list thereof.

        Raises:
            ValueError: if a column explicitly specified (by name, id or index) is not found
            TypeError: if cols are not of requested type
        """
        if isinstance(cols, (str, int)):
            not_found = []
            indexes: list[int] = self.data.safeColIndexList(cast(list[int]|list[str], [ cols ]), not_found=not_found)
            if 0 < len(not_found):
                raise ValueError(f"Column ({not_found[0]}) could not be found!")
            return self.data.colArray(indexes[0])
        elif cols is None or _is_str_list(cols) or _is_int_list(cols) or isinstance(cols, re.Pattern) or callable(cols):
            not_found = []
            indexes: list[int] = self.data.safeColIndexList(cols, not_found=not_found)
            if 0 < len(not_found):
                raise ValueError(f"Some columns ({', '.join(str(c) for c in  not_found)}) could not be found!")
            return self.data.colArray(indexes) if len(indexes) else []
        else:
            raise TypeError("Argument `cols` must be column name, ID or index or a list of these or None, regexp Pattern or Callable.")

    def dataFrame(self,
                *,
                use_col_ids: bool = False,
                no_hidden: bool = False, no_loops: bool = False, no_loop_indexes: bool = False,
                no_bin: bool = False, no_bin_id: bool = False, no_bin_entity: bool = False,
                no_well: bool = False, no_file: bool = False,
                excl: SafeColSpec = None, incl: SafeColSpec = None) -> "pd.DataFrame":
        """Return Pandas DataFrame with all columns except as specified.

        By default all columns are returned.
        If no_* or excl kwarg is present, matching columns are removed.
        If incl is present, matching columns are added (after exclusion)

        Args:
            use_col_ids: columns are indexed by column ids and not by column titles
            no_*: exclude specific system column/hidden columns
            excl: exclude any column
            incl: include any column

        Returns:
            Pandas DataFrame.

        Raises:
            AssertError: if a column explicitly specified column (by name, id or index) is not found
        """

        excl_indexes: list[int] = []
        incl_indexes: list[int] = []
        if no_loops == True:
            excl_indexes += self.data.safeColIndexList(re.compile(r"^(_Loop.+Index)|(_Well)|(_File)$"))
        if no_loop_indexes == True:
            excl_indexes += self.data.safeColIndexList(re.compile(r"^_Loop.+Index$"))
        if no_well == True:
            excl_indexes += self.data.safeColIndexList(re.compile(r"^_Well$"))
        if no_file == True:
            excl_indexes += self.data.safeColIndexList(re.compile(r"^_File$"))
        if no_hidden == True:
            excl_indexes += self.data.safeColIndexList(lambda _, meta: not meta.get("hidden", False))
        if no_bin == True:
            excl_indexes += self.data.safeColIndexList(re.compile(r"^(_ObjId)|(_ObjId3d)|(_Entity)$"))
        if no_bin_id == True:
            excl_indexes += self.data.safeColIndexList(re.compile(r"^(_ObjId)|(_ObjId3d)$"))
        if no_bin_entity == True:
            excl_indexes += self.data.safeColIndexList(re.compile(r"^_Entity$"))
        if excl is not None:
            excl_indexes += self.data.safeColIndexList(excl)
        if incl is not None:
            incl_indexes = self.data.safeColIndexList(incl)
        indexes = (set(self.data.safeColIndexList()) - set(excl_indexes)) | set(incl_indexes)
        return self.data.pandasDataFrame(list(indexes), use_col_ids=use_col_ids)

    def metaDataFrame(self) -> "pd.DataFrame":
        """Return Pandas DataFrame describing input table columns."""
        return self.data.pandasMetaDataFrame()

@dataclass(frozen=True, kw_only=True)
class OutputTableData(ParameterDef, RecordedData):
    """Writable GA3 output table data filled in run(...)."""
    data: LimTableLike|None

    def withColDataFrom(self, *args: Sequence[InputTableData], remaining_cols: list[str]|None = None) -> OutputTableData:
        """Fill matching output columns from input tables or Pandas objects.

        Columns are matched by output column IDs. This is useful after
        makeNew(name, input_table) when most output columns should keep the
        input values and only derived columns need to be filled separately.

        Args:
            *args: InputTableData, pandas.DataFrame, or pandas.Series sources.
            remaining_cols: Optional list that receives output column IDs not filled.

        Returns:
            self
        """
        if not self.data:
            return self
        col_ids = set(self.data.colIdList)
        for arg in args:
            if isinstance(arg, InputTableData):
                ids = col_ids.intersection(set(arg.data.colIdList))
                data = arg.data.colData(list(ids))
                self.data.setColData(list(ids), data)
                col_ids -= ids
            else:
                try:
                    import pandas
                    not_found: list[str] = []
                    if isinstance(arg, pandas.DataFrame):
                        col_ids -= set(self.data.setPandasDataFrame(arg, list(col_ids), not_found=not_found))
                    elif isinstance(arg, pandas.Series):
                        col_ids -= set(self.data.setPandasDataFrame(pandas.DataFrame(arg), list(col_ids), not_found=not_found))
                    else:
                        raise RuntimeError("Unexpected argument type")
                except ImportError:
                    pass
        if isinstance(remaining_cols, list) and len(col_ids):
            remaining_cols += list(col_ids)
        return self

    def withColData(self, cols: str|int|SafeColSpec, data: MultiColData) -> OutputTableData:
        """Fill table with column data and return self.

        Column specifier:
        - column name, ID or index returns single column
        - list of column names, IDs or indexes returns list of columns (must be present)
        - None return all columns
        - re.Pattern return columns where fullmatch succeeded on column ID or Title
        - Callable[[str, dict], bool] returns columns where callable returns True for (ID, Metadata)

        The data must have same length as cols.

        When column names, IDs or indexes are specified explicitly in a list they must
        be present otherwise and AssertError is raised.

        Args:
            cols: col name, id or index or a list there of or a Pattern or Callable
            data: list of values, 1D nd array or a list of these or 2D nd array

        Returns:
            self

        Raises:
            ValueError: if a column explicitly specified column (by name, id or index) is not found
            TypeError: if cols or data are not of requested type
        """
        if not self.data:
            return self
        if isinstance(cols, (str, int)):
            not_found = []
            indexes: list[int] = self.data.safeColIndexList(cast(list[int]|list[str], [ cols ]), not_found=not_found)
            if 0 < len(not_found):
                raise ValueError(f"Column ({not_found[0]}) could not be found!")
            if not (isinstance(data, list) or _is_1d_array(data)):
                raise TypeError("Argument `data` must be a list or a 1D nd array")
            self.data.setColData(indexes, cast(MultiColData, [ data ]))
            return self
        elif cols is None or _is_str_list(cols) or _is_int_list(cols) or isinstance(cols, re.Pattern) or callable(cols):
            not_found = []
            indexes: list[int] = self.data.safeColIndexList(cols, not_found=not_found)
            if 0 < len(not_found):
                raise ValueError(f"Some columns ({', '.join(str(c) for c in  not_found)}) could not be found!")
            if not (_is_list_list(data) or _is_1d_array_list(data) or _is_2d_array(data)):
                raise TypeError("Argument `data` must be a list of lists, list of 1D nd arrays or a 2D nd array")
            if (isinstance(data, list) and len(indexes) != len(data)) or (getattr(data, '__array__', False) and len(indexes) != cast(ArrayLike, data).shape[0]):
                raise ValueError("Arguments `cols` and `data` must have same length")
            self.data.setColData(indexes, cast(MultiColData, data))
            return self
        else:
            raise TypeError("Argument `cols` must be column name, ID or index or a list of these or None, regexp Pattern or Callable.")

    def dataFrame(self, *, use_col_ids: bool = False) -> "pd.DataFrame":
        """Return Pandas DataFrame with all columns.

        Args:
            use_col_ids: columns are indexed by column ids and not by column titles

        Returns:
            Pandas DataFrame.
        """

        return self.data.pandasDataFrame(use_col_ids=use_col_ids)

    def withDataFrame(self, df: "pd.DataFrame", cols: str|int|SafeColSpec = None) -> OutputTableData:
        """Fill table with Pandas DataFrame.

        Column specifier:
        - column name, ID or index returns single column
        - list of column names, IDs or indexes returns list of columns (must be present)
        - None return all columns
        - re.Pattern return columns where fullmatch succeeded on column ID or Title
        - Callable[[str, dict], bool] returns columns where callable returns True for (ID, Metadata)

        When column names, IDs or indexes are specified explicitly in a list they must
        be present otherwise and AssertError is raised.

        Args:
            df: Pandas DataFrame
            cols: col name, id or index or a list there of or a Pattern or Callable

        Returns:
            self

        Raises:
            ValueError: if a column explicitly specified (by name, id or index) is not found
            TypeError: if cols or data are not of requested type
        """
        if not self.data:
            return self
        if isinstance(cols, (str, int)):
            not_found = []
            indexes: list[int] = self.data.safeColIndexList(cast(list[int]|list[str], [ cols ]), not_found=not_found)
            if 0 < len(not_found):
                raise ValueError(f"Column ({not_found[0]}) could not be found!")
            self.data.setPandasDataFrame(df, indexes, not_found=not_found)
            if 0 < len(not_found):
                raise ValueError(f"Column ({not_found[0]}) could not be in the Dataframe df!")
            return self
        elif cols is None or _is_str_list(cols) or _is_int_list(cols) or isinstance(cols, re.Pattern) or callable(cols):
            not_found = []
            indexes: list[int] = self.data.safeColIndexList(cols, not_found=not_found)
            if 0 < len(not_found):
                raise ValueError(f"Some columns ({', '.join(str(c) for c in  not_found)}) could not be found!")
            self.data.setPandasDataFrame(df, indexes, not_found=not_found)
            if 0 < len(not_found):
                raise ValueError(f"Some columns ({', '.join(str(c) for c in  not_found)}) could not be in the Dataframe df!")
            return self
        else:
            raise TypeError("Argument `cols` must be column name, ID or index or a list of these or None, regexp Pattern or Callable.")

    def copyRecordedDataFrom(self, other: RecordedData) -> None:
        self._copyRecordedDataFrom(other)

    def copyFrom(self, other: RecordedData) -> None:
        assert isinstance(other, OutputTableData), "Unexpected argument type"
        if self.data and other.data:
            col_ids = other.data.colIdList
            col_data = other.data.colData(col_ids)
            self.data.setColData(col_ids, col_data)
            self.copyRecordedDataFrom(other)

    def withMplImage(self, loopCoordinates: tuple[int], plt: MplFigureLike|tuple[MplFigureLike,MplFigureLike],
                        *, name: str|None = None, iconres: str|None = None,
                        objectFit: str|None = None, tableRowVisibility: str|None = None,
                        savefig_kwargs: dict = { 'format': 'png' }
                        ) -> OutputTableData:
        """Store a matplotlib figure in a result table.

        The output table must be defined with OutputTableDef.makeResult(...).
        Passing a tuple stores separate light-theme and dark-theme images.

        Args:
            loopCoordinates: Usually ctx.outCoordinates for the current result row.
            plt: Matplotlib figure/canvas, or (light, dark) figure/canvas pair.
            name: Optional result tab name.
            iconres: Optional GA3 icon resource name.
            objectFit: Optional CSS object-fit value for the image view.
            tableRowVisibility: Optional result-row visibility mode.
            savefig_kwargs: Keyword arguments used when plt.savefig(...) is called.

        Returns:
            self
        """
        assert isinstance(loopCoordinates, tuple), "Unexpected argument type"
        if self.data is None:
            return self
        def save(plt: MplFigureLike, whatErrStr: str|None) -> bytes:
            import io
            buf = io.BytesIO()
            if hasattr(plt, 'print_png'):
                plt.print_png(buf)
            elif hasattr(plt, 'savefig'):
                plt.savefig(buf, **savefig_kwargs)
            else:
                raise RuntimeError(f"{whatErrStr} is of unexpected type")
            buf.seek(0)
            return buf.getvalue()

        def createPrivateTable(my_data: LimTableLike, loopCoordinates: tuple[int], mimeType: str, imageDataList: list[tuple[str, str, str, bytes]]) -> LimTableLike:
            import base64, uuid
            from limtabledata import LimTableDataBase
            imgCols = []
            for id_, feature, title, data in imageDataList:
                imgCols.append((id_, [ base64.b64encode(data).decode('ascii') ], dict(decltype='QString', title=title, feature=feature, mimeType=mimeType)))
            metadata = dict(_systemFlags=['result', 'table'], contains=my_data.tableMetadata['contains'], uniqueStamp=str(uuid.uuid4()), parameterUuid=str(uuid.uuid4()))
            if 'accumulatedLoops' in my_data.tableMetadata:
                metadata['accumulatedLoops'] = my_data.tableMetadata['accumulatedLoops']
            loopCols = [(f"_loop{loop.name}Index",  [ loopCoordinates[i] + 1 ], dict(decltype="int", title=f"{loop.name}Index", loopName=loop.name, loopType=int(loop.type))) for i, loop in enumerate(self.loopDefList)]
            return LimTableDataBase(*(loopCols + imgCols), tabName="a", tabMetadata=metadata) # type: ignore

        if isinstance(plt, tuple) and len(plt) == 2:
            data = [
                (f"{self.parentNode.uuid}.imageDataLight", "imageDataLight", "ImageDataLight", save(plt[0], "plt[0]")),
                (f"{self.parentNode.uuid}.imageDataDark", "imageDataDark", "ImageDataDark", save(plt[1], "plt[1]"))]
        else:
            data = [ (f"{self.parentNode.uuid}.imageData", "imageData", "ImageData", save(plt, "plt")) ]

        state = dict(
            name = 'Mpl graph' if name is not None else name,
            iconres = 'line_common' if iconres is None else iconres,
            tableRowVisibility = 'currentFrame' if tableRowVisibility is None else tableRowVisibility)
        if objectFit is not None:
            state['objectFit'] = objectFit

        a = createPrivateTable(self.data, loopCoordinates, "image/png", data)
        self._makeResult(a, "LimResultImageView", state)
        return self

    def _makeResult(self, data: LimTableLike, jsClassName: str, state: dict[str, Any] = {}) -> OutputTableData:
        if self.data is None:
            return self
        from limtabledata import LimTableDataBase
        assert isinstance(data, LimTableDataBase) and isinstance(jsClassName, str) and isinstance(state, dict), "Unexpected argument type"
        import json, uuid
        privateTable = LimTableDataBase(data)
        privateTable.setTableName("a")
        state["_tableName"] = privateTable.tableName
        state["_nodeUuid"] = self.parentNode.uuid
        state.setdefault("_tableParamUuid", str(uuid.uuid4()))
        initialState = dict(panes=[dict(className='LimPane', state=dict(tabs=[dict(className=jsClassName, state=state)]))])
        self.withColData([ "Html", "InitialState" ], [ [ "" ], [ json.dumps(initialState) ] ])
        self.data.insertPrivateTable(cast(LimTableLike, privateTable))
        return self

@dataclass
class Program:
    """GA3 invocation plan returned by build(...) to select loops handled in Python."""
    allLoopDefs: list[LoopDef]
    overLoops: list[str] = field(default_factory=list, init=False)

    def overZStack(self) -> Program:
        """Run the program over Z-stack loops instead of per-Z invocation."""
        for loop in self.allLoopDefs:
            if loop.type == ExperimentLoopType.loopTypeZStack and loop.name not in self.overLoops:
                self.overLoops.append(loop.name)
        return self

    def overTimeLapse(self) -> Program:
        """Run the program over time-lapse loops instead of per-time invocation."""
        for loop in self.allLoopDefs:
            if loop.type == ExperimentLoopType.loopTypeTLapse and loop.name not in self.overLoops:
                self.overLoops.append(loop.name)
        return self

    def overMultiPoint(self) -> Program:
        """Run the program over multipoint/site loops instead of per-site invocation."""
        for loop in self.allLoopDefs:
            if loop.type == ExperimentLoopType.loopTypeXYSite and loop.name not in self.overLoops:
                self.overLoops.append(loop.name)
        return self

    def overAll(self) -> Program:
        """Run the program over all available loops."""
        self.overLoops = [loop.name for loop in self.allLoopDefs]
        return self

AnyInDef: TypeAlias = InputChannelDef|InputBinaryDef|InputTableDef|None
AnyOutDef: TypeAlias = OutputChannelDef|OutputBinaryDef|OutputTableDef
AnyInData: TypeAlias = InputChannelData|InputBinaryData|InputTableData|None
AnyOutData: TypeAlias = OutputChannelData|OutputBinaryData|OutputTableData

InDefTuple: TypeAlias = tuple[AnyInDef, ...]
OutDefTuple: TypeAlias = tuple[AnyOutDef, ...]
UserParTuple: TypeAlias = tuple[int|float|str, ...]
LoopDefs: TypeAlias =  list[LoopDef]
InDataTuple: TypeAlias = tuple[AnyInData, ...]
OutDataTuple: TypeAlias = tuple[AnyOutData, ...]

@dataclass
class ReduceProgram(Program):
    """GA3 reduction program that accumulates selected loops before final output."""
    pass

@dataclass
class TwoPassProgram(Program):
    """GA3 two-pass program for algorithms that need a scan pass before output."""
    pass

@dataclass(kw_only=True)
class RunContext:
    """GA3 execution context for the current run(...) invocation."""
    inpFilename: str
    outFilename: str
    inpParameterCoordinates: tuple[tuple[int]]
    outCoordinates: tuple[int]
    programLoopsOver: list[str]|None          # loop names
    programCoordinateIndexes: tuple[int]|None # innerloops: indexes of loops that must be handled in Python (not by GA3) e.g. Z-Stack loop
    programPass: int
    programIndex: int
    programLength: int
    finalCall: bool
    tempPath: str

    @property
    def shouldAbort(self) -> bool:
        """Return True when GA3 has requested cancellation of the current run."""
        try:
            return self._should_abort() # type: ignore
        except AttributeError:
            return False

def _parse_color(color: AnyColorDef) -> int:
    if type(color) == int:
        return color & 0xFFFFFF
    elif type(color) == str and len(color) == 7 and color[0] == '#':
        return int(color[1:], 16)
    elif type(color) == tuple:
        assert 3 <= len(color), "Unexpected length of argument 'color'"
        return (color[0] << 16) + (color[1] << 8) + color[2]
    else:
        return 0

###############################################################################
## Out-process
###############################################################################

@dataclass
class BaseRequest:
    sto: tuple[str, int]

@dataclass
class OutputRequest(BaseRequest):
    inp: tuple[AnyInDef, ...]
    out: tuple[AnyOutDef, ...]
    par: tuple[int|float|str, ...]

@dataclass
class BuildRequest(BaseRequest):
    par: tuple[int|float|str, ...]
    loops: list[LoopDef]

@dataclass
class RunRequest(BaseRequest):
    inp: tuple[AnyInData, ...]
    out: tuple[AnyOutData, ...]
    par: tuple[int|float|str, ...]
    ctx: RunContext
    shm: dict[str, dict]

class ChildProcessError(Exception):
    def __str__(self) -> str:
        return "\n" + self.args[0]

    def __repr__(self) -> str:
        return "\n" + self.args[0]

class CalledProcessError(subprocess.CalledProcessError):
    def __repr__(self) -> str:
        if 0 == self.returncode:
            return """
The child process exited too early!

Check that you have the following lines at the end of your code:

if __name__ == '__main__':
    from limnode import print
    limnode.child_main(run, output, build)

"""
        str_repr = "Exited code %d." % (self.returncode)
        if self.cmd:
            str_repr += "\n  From: %s" % (self.cmd)
        if self.stdout:
            str_repr += "\n%s" % (self.stdout.decode())
        if self.stderr:
            str_repr += "\n%s" % (self.stderr.decode())
        return str_repr

class StdoutToSharedMemoryWriter:
    def __init__(self, shm_name, size):
        from threading import Lock
        from multiprocessing.shared_memory import SharedMemory
        self.shm = SharedMemory(name=shm_name)
        assert self.shm.buf is not None, "Stdout: Failed to open shared memory"
        self.buf = self.shm.buf
        self.size = size
        self.lock = Lock()
        self.offset = 0

    def write(self, data):
        with self.lock:
            encoded = data.encode('utf-8')
            n = len(encoded)
            if self.offset + n > self.size:
                return
            self.buf[self.offset:self.offset+n] = encoded
            self.offset += n

    def flush(self):
        pass  # Required for file-like compatibility

    def __enter__(self):
        global _nis_shm_wtiter
        _nis_shm_wtiter = self

    def __exit__(self, exc_type, exc_value, traceback):
        global _nis_shm_wtiter
        _nis_shm_wtiter = None

class SharedMemoryForStdout:
    def __init__(self, size=10*1024*1024):
        self.size = size
        self.shm = None

    def __enter__(self):
        from multiprocessing.shared_memory import SharedMemory
        self.shm = SharedMemory(create=True, size=self.size)
        assert self.shm.buf is not None, "Stdout: Failed to create shared memory"
        self.shm.buf[:self.size] = b'\x00' * self.size
        return (self.shm.name, self.size)

    def __exit__(self, exc_type, exc_value, traceback):
        assert self.shm and (self.shm.buf is not None), "Stdout: Shared memory not present"
        data = bytes(self.shm.buf[:]).rstrip(b'\x00')
        if len(data):
            sys.stdout.write(data.decode('utf-8'))
            sys.stdout.flush()
        self.shm.close()

_builtin_print = print
_nis_shm_wtiter = None
def print(*args, **kwargs):
    """Write to the GA3 Python output stream from managed child execution.

    Import this instead of builtins.print inside scripts that run through
    child_main(...). The text is forwarded to the host process output.
    """
    global _nis_shm_wtiter
    if (_nis_shm_wtiter is None):
        return
    _nis_shm_wtiter.write(kwargs.get("sep", " ").join([str(arg) for arg in args]) + kwargs.get("end", "\n"))
    _nis_shm_wtiter.flush()

_nis_python_logfile = None
def log(*args, **kwargs):
    """Append diagnostic text to the NIS Python log when NIS_LOG_PATH is set."""
    global _nis_python_logfile
    import datetime
    if _nis_python_logfile is None and "NIS_LOG_PATH" in os.environ:
        today = datetime.datetime.now()
        _nis_python_logfile = open(f"{os.environ['NIS_LOG_PATH']}\\pylog_{today.year}-{today.month:02d}-{today.day:02d}.txt", "a")
    if _nis_python_logfile is not None:
        if kwargs.get("prefix", True):
            args = (datetime.datetime.now().isoformat(), ) + args
        _nis_python_logfile.write(kwargs.get("sep", " ").join([str(arg) for arg in args]) + kwargs.get("end", "\n"))
        if kwargs.get("flush", True):
            _nis_python_logfile.flush()


class ParentNode:
    def __init__(self, module_name: str|None = None) -> None:
        self._module_name: str = module_name if module_name is not None else "???"
        self._child_process: subprocess.Popen|None = None
        self._parent_connection: "multiprocessing.connection.PipeConnection|None" = None
        self._lock = threading_RLock()

    def __del__(self):
        self.shutdown_child_process()
 
    @classmethod
    def job_for_child_processes(cls):
        """Create a job object that kills all child processes when parent dies"""
        if (job := getattr(cls, "_job", None)) is not None:
            return job
        
        import win32job
        
        # Create a job object
        job = win32job.CreateJobObject(None, "")
        
        # Configure the job to kill all processes when the handle is closed
        extended_info = win32job.QueryInformationJobObject(
            job, 
            win32job.JobObjectExtendedLimitInformation
        )
        extended_info['BasicLimitInformation']['LimitFlags'] = (
            win32job.JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE
        )
        win32job.SetInformationJobObject(
            job,
            win32job.JobObjectExtendedLimitInformation,
            extended_info
        )
        
        setattr(cls, "_job", job)
        return job

    @locked
    def ensure_child_process_alive(self, executable: str, child_code: str) -> None:
        if isinstance(self._child_process, subprocess.Popen) and self._parent_connection is not None:
            try:
                _builtin_print(f"{self._module_name}: Checking if child process is alive...")
                self._child_process.communicate(timeout=0)
                _builtin_print(f"{self._module_name}: Child process terminated with returncode={self._child_process.returncode}")
            except subprocess.TimeoutExpired as _:
                _builtin_print(f"{self._module_name}: Child process is still alive")
                return
        self.start_child_process(executable, child_code)

    @locked
    def shutdown_child_process(self) -> None:
        if self._child_process is None:
            return
        self._child_process.kill()
        _builtin_print(f"{self._module_name}: Waiting for child process to finish...")
        self._child_process.communicate()
        _builtin_print(f"{self._module_name}: Child finished!")
        self._child_process, self._parent_connection = None, None

    # 1. Create a special pipe for SYNCHRONIZATION
    # 2. Create the child process with Popen("pythonw.exe - NNN") where NNN is the FILENO of the special sync pipe and - indicate to python the code to be executed will be passed on stdin
    # 3. Send the code to child, don't wait for termination (timeout)
    # 4. Pool sync pipe until READY received
    @locked
    def start_child_process(self, executable: str, child_code: str) -> None:
        import win32api, win32con, win32job
        from multiprocessing import Pipe
        parent_connection, child_connection = Pipe()

        python_home = os.path.dirname(executable)
        env = os.environ.copy()
        env["PYTHONNOUSERSITE"] = "1"
        env["PYTHONHOME"] = python_home
        if "PYTHONPATH" in env:
            env.pop("PYTHONPATH")
        env["PATH"] = f"{python_home};{env['PATH']}"

        _builtin_print(f"{self._module_name}: Creating child process...")
        child_process = subprocess.Popen(
            [executable, '-', str(child_connection.fileno())], 
            stdin=subprocess.PIPE, 
            stdout=subprocess.PIPE, 
            stderr=subprocess.PIPE, env=env
        )
        
        # Assign the process to the job
        process_handle = win32api.OpenProcess(
            win32con.PROCESS_ALL_ACCESS,
            False,
            child_process.pid
        )
        job = ParentNode.job_for_child_processes()
        win32job.AssignProcessToJobObject(job, process_handle)

        _builtin_print(f"{self._module_name}: Passing the script to the child process...")
        try:
            stdout, stderr = child_process.communicate(child_code.encode(), timeout=0)
            raise CalledProcessError(child_process.returncode, executable, stdout, stderr)

        except subprocess.TimeoutExpired as _:
            pass

        _builtin_print(f"{self._module_name}: Waiting for the child process to be ready...")
        while not parent_connection.poll(0.1):
            try:
                stdout, stderr = child_process.communicate(timeout=0)
                raise CalledProcessError(child_process.returncode, executable, stdout, stderr)

            except subprocess.TimeoutExpired as _:
                pass

        if (parent_connection is None) or (parent_connection.recv() != "CHILD PROCESS READY"):
            raise RuntimeError(f"{self._module_name}: Child process not ready")

        _builtin_print(f"{self._module_name}: Child process ready and waiting for data!")
        child_connection._handle = None # type: ignore

        self._child_process, self._parent_connection = child_process, parent_connection

    @locked
    def parent_output(self, inp: tuple[AnyInDef, ...], out: tuple[AnyOutDef, ...], par: tuple[int|float|str, ...]) -> None:
        assert self._parent_connection is not None, f"{self._module_name}: Connection must not be None"
        with SharedMemoryForStdout() as stdout_shm_name_and_size:
            # send request to the child
            self._parent_connection.send(OutputRequest(inp=inp, out=out, par=par, sto=stdout_shm_name_and_size))
            # wait for a response from the child process
            ret = self._parent_connection.recv()

        if isinstance(ret, Exception):
            raise ret
        else:
            for out_item, ret_item in zip(out, ret):
                out_item.copyFrom(ret_item)

    @locked
    def parent_build(self, par: tuple[int|float|str, ...], loops: list[LoopDef]) -> Program|None:
        assert self._parent_connection is not None, f"{self._module_name}: Connection must not be None"
        with SharedMemoryForStdout() as stdout_shm_name_and_size:
            # send request to the child
            self._parent_connection.send(BuildRequest(par=par, loops=loops, sto=stdout_shm_name_and_size))
            # wait for a response from the child process
            ret = self._parent_connection.recv()

        if isinstance(ret, Exception):
            raise ret
        else:
            return ret

    @locked
    def parent_run(self, inp: tuple[AnyInData, ...], out: tuple[AnyOutData, ...], par: tuple[int|float|str, ...], ctx: RunContext) -> None:
        global _child_process, _parent_connection
        import numpy as np
        from multiprocessing.shared_memory import SharedMemory
        par_sh_mem, shm = {}, {}
        src_arrays = {}
        for item in (inp + out):
            if isinstance(item, ArrayData) and item.data is not None:
                src_array = cast(np.ndarray, item.data)
                src_arrays[item.parName] = src_array
                item._setData(None)
                sh_mem = SharedMemory(create=True, size=src_array.nbytes)
                tmp = np.ndarray(src_array.shape, src_array.dtype, buffer=sh_mem.buf)
                tmp[:] = src_array
                shm[item.parName] = dict(name=sh_mem.name, shape=src_array.shape, dtype=src_array.dtype, writeable=src_array.flags.writeable)
                par_sh_mem[item.parName] = sh_mem

        with SharedMemoryForStdout() as stdout_shm_name_and_size:
            # send request to the child
            assert self._child_process is not None, f"{self._module_name}: Child process must not be None"
            assert self._parent_connection is not None, f"{self._module_name}: Connection must not be None"
            self._parent_connection.send(RunRequest(inp=inp, out=out, par=par, ctx=RunContext(**asdict(ctx)), shm=shm, sto=stdout_shm_name_and_size))

            # wait for a response from the child process
            while not self._parent_connection.poll(0.1):
                if ctx.shouldAbort:
                    _builtin_print(f"{self._module_name}: Aborting - killing child process abd waiting...")
                    self.shutdown_child_process()
                    return

            # can raise EOFError
            ret = self._parent_connection.recv()
            if isinstance(ret, Exception):
                raise ret

        for item in inp:
            if isinstance(item, ArrayData) and item.parName in par_sh_mem:
                sh_mem = par_sh_mem[item.parName]
                item._setData(src_arrays.pop(item.parName, None))
                sh_mem.close()

        for item, ret_item in zip(out, ret):
            if isinstance(item, ArrayData) and item.parName in par_sh_mem:
                sh_mem = par_sh_mem[item.parName]
                src_array = src_arrays.pop(item.parName, None)
                tmp = np.ndarray(shm[item.parName]['shape'], dtype=np.dtype(shm[item.parName]['dtype']), buffer=sh_mem.buf)
                np.copyto(src_array, tmp, casting="no")
                item._setData(src_array)
                del tmp
                sh_mem.close()
            elif isinstance(item, OutputTableData):
                item.copyFrom(ret_item)


def child_main(
        run: Callable[[tuple[AnyInData, ...], tuple[AnyOutData, ...], tuple[int|float|str, ...], RunContext], Exception|None],
        output: Callable[[tuple[AnyInDef, ...], tuple[AnyOutDef, ...], tuple[int|float|str, ...]], None]|None = None,
        build: Callable[[tuple[int|float|str, ...], list[LoopDef]], Exception|None]|None = None,
    ) -> None:
    """Run the GA3 child-process event loop for a Python node script.

    Put this at the end of authored scripts:

        if __name__ == "__main__":
            from limnode import print
            limnode.child_main(run, output, build)

    In GA3, this connects the script to the host process. Outside GA3, it
    replays pickled development inputs from the local data folder.

    Args:
        run: Runtime callback that fills outputs.
        output: Optional callback that defines output metadata and schemas.
        build: Optional callback that returns Program, ReduceProgram, TwoPassProgram, or None.
    """
    import os, sys, traceback
    from multiprocessing.connection import PipeConnection
    from multiprocessing.reduction import steal_handle

    global dll_dir_handles
    dll_dir_handles = []

    # Since mamba activate is not called add 'Library\bin' to the PATH manually
    dir_library_bin = os.path.join(os.path.dirname(sys.executable), "Library", "bin")
    if os.path.isdir(dir_library_bin):
        dll_dir_handles.append(os.add_dll_directory(dir_library_bin))
        os.environ["PATH"] = f"{dir_library_bin};{os.environ.get('PATH', '')}"

    # Since mamba activate is not called add 'bin' to the PATH manually (may be used by cuda)
    dir_bin = os.path.join(os.path.dirname(sys.executable), "bin")
    if os.path.isdir(dir_bin):
        dll_dir_handles.append(os.add_dll_directory(dir_bin))
        os.environ["PATH"] = f"{dir_bin};{os.environ.get('PATH', '')}"

    if os.environ.get("NIS_APP_NAME") is None:
        outside_main(run, output, build)
        return

    assert (3, 10) <= sys.version_info, "Python version too low (must be at least 3.10)"

    connection = PipeConnection(steal_handle(os.getppid(), int(sys.argv[1])))
    connection.send("CHILD PROCESS READY")
    while True:
        try:
            request = connection.recv()
        except Exception as e:
            connection.send(ChildProcessError("".join(traceback.format_exception(e))))
            continue

        with StdoutToSharedMemoryWriter(*request.sto):
            if isinstance(request, OutputRequest) and output is not None:
                try:
                    output(request.inp, request.out, request.par)
                except Exception as e:
                    connection.send(ChildProcessError("".join(traceback.format_exception(e))))
                else:
                    connection.send(request.out)
            elif isinstance(request, BuildRequest) and build is not None:
                try:
                    ret = build(request.par, request.loops)
                except Exception as e:
                    connection.send(ChildProcessError("".join(traceback.format_exception(e))))
                else:
                    connection.send(ret)
            elif isinstance(request, RunRequest) and run is not None:
                try:
                    _child_run(run, request.inp, request.out, request.par, request.ctx, request.shm)
                except Exception as e:
                    if (type(e).__name__ == "Ga3Error"):
                        connection.send(Ga3Error("".join(traceback.format_exception(e))))
                    else:
                        connection.send(ChildProcessError("".join(traceback.format_exception(e))))
                else:
                    connection.send(request.out)
            else:
                break

def _child_run(
        node_run: Callable[[tuple[AnyInData, ...], tuple[AnyOutData, ...], tuple[int|float|str, ...], RunContext], Exception|None],
        inp: tuple[AnyInData, ...], out: tuple[AnyOutData, ...], par: tuple[int|float|str, ...],
        ctx: RunContext,
        shm: dict
    ) -> Exception|None:
    import numpy as np
    from multiprocessing.shared_memory import SharedMemory
    par_sh_mem = {}
    for item in (inp + out):
        if isinstance(item, ArrayData) and item.parName in shm:
            sh_mem = SharedMemory(name=shm[item.parName]['name'], create=False)
            tmp = np.ndarray(shm[item.parName]['shape'], shm[item.parName]['dtype'], buffer=sh_mem.buf)
            tmp.flags.writeable = shm[item.parName]['writeable']
            item._setData(cast(ArrayLike, tmp))
            par_sh_mem[item.parName] = sh_mem

    try:
        node_run(inp, out, par, ctx)

    except Exception as _:
        raise

    finally:
        for item in (inp + out):
            if isinstance(item, ArrayData) and item.parName in par_sh_mem:
                item._setData(None)
                par_sh_mem[item.parName].close()

###############################################################################
## Develop in VS Code
###############################################################################

def outside_output(folder: str, inp: tuple[AnyInDef, ...], out: tuple[AnyOutDef, ...], par: tuple[int|float|str, ...]) -> None:
    """Write output(...) inputs for local development replay."""
    import pickle
    with open(os.path.join(folder, "data", "output_inp.pkl"), "wb") as f:
        pickle.dump(inp, f)
    with open(os.path.join(folder, "data", "output_out.pkl"), "wb") as f:
        pickle.dump(out, f)
    with open(os.path.join(folder, "data", "output_par.pkl"), "wb") as f:
        pickle.dump(par, f)

def outside_build(folder: str, par: tuple[int|float|str, ...], loops: list[LoopDef]) -> Program|None:
    """Write build(...) inputs for local development replay."""
    import pickle
    with open(os.path.join(folder, "data", "build_par.pkl"), "wb") as f:
        pickle.dump(par, f)
    with open(os.path.join(folder, "data", "build_loops.pkl"), "wb") as f:
        pickle.dump(loops, f)

def outside_run(folder: str, inp: tuple[AnyInData, ...], out: tuple[AnyOutData, ...], par: tuple[int|float|str, ...], ctx: RunContext) -> None:
    """Write run(...) inputs for local development replay."""
    import pickle
    out_path = os.path.join(folder, "data", "run_out.pkl")
    inp_path = os.path.join(folder, "data", "run_inp.pkl")
    if (not os.path.exists(inp_path)) or os.path.getmtime(out_path) <= os.path.getmtime(inp_path):
        with open(out_path, "wb") as f:
            pickle.dump(out, f)
        with open(inp_path, "wb") as f:
            pickle.dump(inp, f)
        with open(os.path.join(folder, "data", "run_par.pkl"), "wb") as f:
            pickle.dump(par, f)
        with open(os.path.join(folder, "data", "run_ctx.pkl"), "wb") as f:
            pickle.dump(ctx, f)

def outside_main(run, output, build) -> None:
    import pickle
    with open(os.path.join("data", "output_inp.pkl"), "rb") as f:
        inp = pickle.load(f)
    with open(os.path.join("data", "output_out.pkl"), "rb") as f:
        out = pickle.load(f)
    with open(os.path.join("data", "output_par.pkl"), "rb") as f:
        par = pickle.load(f)
    output(inp, out, par)

    #build()

    with open(os.path.join("data", "run_inp.pkl"), "rb") as f:
        inp = pickle.load(f)
    with open(os.path.join("data", "run_out.pkl"), "rb") as f:
        out = pickle.load(f)
    with open(os.path.join("data", "run_par.pkl"), "rb") as f:
        par = pickle.load(f)
    with open(os.path.join("data", "run_ctx.pkl"), "rb") as f:
        ctx = pickle.load(f)
    run(inp, out, par, ctx)

###############################################################################
## Helper functions
###############################################################################

def separateLabeledImage(src: ArrayLike) -> ArrayLike:
    """Separate touching labels in a 2D labeled image.

    Pixels that touch a higher-valued label in their 3x3 neighborhood are set
    to zero. Use this to create a one-pixel background gap between adjacent
    labeled objects before measurement or display.

    Args:
        src: 2D numpy ndarray of integer labels.

    Returns:
        2D numpy ndarray with boundary pixels cleared.
    """
    import numpy as np
    from numpy.lib.stride_tricks import as_strided

    if not isinstance(src, np.ndarray):
        raise TypeError(f"Expected numpy ndarray, got {type(src)}")
    if src.ndim != 2:
        raise TypeError(f"Expected 2-dimensional array, got ndim={src.ndim}")

    padded = np.pad(src, (1,1))
    shape = (src.shape[0], src.shape[1], 3, 3)
    strides = padded.strides + padded.strides

    windows = as_strided(padded, shape=shape, strides=strides)
    max_values = np.max(windows, axis=(2, 3))
    return cast(ArrayLike, np.where(src < max_values, 0, src))

def separateLabeledImage3d(src: ArrayLike) -> ArrayLike:
    """Separate touching labels in a 3D labeled image.

    Keeps the largest remaining connected component for each original label and
    clears voxels that touch a higher-valued label in their 3x3x3 neighborhood.

    Args:
        src: 3D numpy ndarray of integer labels.

    Returns:
        3D numpy ndarray with boundary voxels cleared.
    """
    import numpy as np
    if not isinstance(src, np.ndarray):
        raise TypeError(f"Expected numpy ndarray, got {type(src)}")
    if src.ndim != 3:
        raise TypeError(f"Expected 3-dimensional array, got ndim={src.ndim}")

    from scipy import ndimage as ndi

    dilated = ndi.grey_dilation(src, footprint=np.ones((3,3,3), dtype=bool))
    separated = np.where(dilated == src, src, 0)

    objects, n = ndi.label(separated) # type: ignore
    object_voxel_counts = np.bincount(objects.ravel())
    object_labels, object_starts = np.unique(objects.ravel(), return_index=True)
    object_to_original = separated.ravel()[object_starts]

    picked_objects = {}
    for label, original in zip(object_labels, object_to_original):
        if label == 0:
            continue
        if original not in picked_objects:
            picked_objects[original] = label
        elif object_voxel_counts[label] > object_voxel_counts[picked_objects[original]]:
            object_to_original[picked_objects[original]] = 0
            picked_objects[original] = label
        else:
            object_to_original[label] = 0

    return object_to_original[objects] # type: ignore