Client

penvm.client.config

Collection of configuration support, mostly used in penvm.client.world.

BaseConfig

Base configuration.

Provides standard methods: add, get, update.

Source code in penvm/src/client/penvm/client/config.py

class BaseConfig:
    """Base configuration.

    Provides standard methods: `add`, `get`, `update`."""

    def __init__(self, world: "WorldConfig", name: str, config: Union[dict, None] = None):
        """Initialize.

        Args:
            world: World configuration object.
            name: Configuration name.
            config: Configuration.
        """
        self.world = world
        self.config = {"name": name}
        if config:
            for k, v in config.items():
                self.add(k, v)

    def add(self, k: Any, v: Any):
        """Add key+value.

        Args:
            k: Key.
            v: Value.
        """
        if k == "targets":
            targets = listify_targets(v)
            v = " ".join(targets)
        self.config[k] = v

    def get(self, k: Any, default: Any = None) -> Any:
        """Get value for key.

        Args:
            k: Key.
            default: Default value if key is not found.

        Returns:
            Value for key.
        """
        return self.config.get(k, default)

    def update(self, config: dict):
        """Update configuration.

        Args:
            config: Configuration.
        """
        self.config.update(config)

__init__(world, name, config=None)

Initialize.

Parameters:

Name	Type	Description	Default
world	WorldConfig	World configuration object.	required
name	str	Configuration name.	required
config	Union[dict, None]	Configuration.	None

Source code in penvm/src/client/penvm/client/config.py

def __init__(self, world: "WorldConfig", name: str, config: Union[dict, None] = None):
    """Initialize.

    Args:
        world: World configuration object.
        name: Configuration name.
        config: Configuration.
    """
    self.world = world
    self.config = {"name": name}
    if config:
        for k, v in config.items():
            self.add(k, v)

add(k, v)

Add key+value.

Parameters:

Name	Type	Description	Default
k	Any	Key.	required
v	Any	Value.	required

Source code in penvm/src/client/penvm/client/config.py

def add(self, k: Any, v: Any):
    """Add key+value.

    Args:
        k: Key.
        v: Value.
    """
    if k == "targets":
        targets = listify_targets(v)
        v = " ".join(targets)
    self.config[k] = v

get(k, default=None)

Get value for key.

Parameters:

Name	Type	Description	Default
k	Any	Key.	required
default	Any	Default value if key is not found.	None

Returns:

Type	Description
Any	Value for key.

Source code in penvm/src/client/penvm/client/config.py

def get(self, k: Any, default: Any = None) -> Any:
    """Get value for key.

    Args:
        k: Key.
        default: Default value if key is not found.

    Returns:
        Value for key.
    """
    return self.config.get(k, default)

update(config)

Update configuration.

Parameters:

Name	Type	Description	Default
config	dict	Configuration.	required

Source code in penvm/src/client/penvm/client/config.py

def update(self, config: dict):
    """Update configuration.

    Args:
        config: Configuration.
    """
    self.config.update(config)

GroupConfig

Bases: BaseConfig

Group configuration.

Source code in penvm/src/client/penvm/client/config.py

class GroupConfig(BaseConfig):
    """Group configuration."""

    def __repr__(self):
        return f"<GroupConfig name={self.config['name']} config={self.config}>"

    def get_targets(self) -> List["TargetConfig"]:
        """Get target configuration objects.

        Returns:
            List of Target configuration objects.
        """
        targets = []
        for target in listify_targets(self.config.get("targets", "")):
            targets.extend(target)
        return targets

get_targets()

Get target configuration objects.

Returns:

Type	Description
List[TargetConfig]	List of Target configuration objects.

Source code in penvm/src/client/penvm/client/config.py

def get_targets(self) -> List["TargetConfig"]:
    """Get target configuration objects.

    Returns:
        List of Target configuration objects.
    """
    targets = []
    for target in listify_targets(self.config.get("targets", "")):
        targets.extend(target)
    return targets

NetworkConfig

Bases: BaseConfig

Network configuration.

Source code in penvm/src/client/penvm/client/config.py

class NetworkConfig(BaseConfig):
    """Network configuration."""

    def __repr__(self):
        return f"<NetworkConfig name={self.config['name']} config={self.config}>"

    def get_targets(self) -> List["TargetConfig"]:
        """Get target configuration objects.

        Returns:
            List of Target configuration objects.
        """
        targets = []
        for target in listify_targets(self.config.get("targets", "")):
            if target.startswith("@"):
                group = self.world.groups.get(target[1:])
                if group:
                    targets.extend(group.get_targets())
            else:
                targets.append(target)
        return targets

get_targets()

Get target configuration objects.

Returns:

Type	Description
List[TargetConfig]	List of Target configuration objects.

Source code in penvm/src/client/penvm/client/config.py

def get_targets(self) -> List["TargetConfig"]:
    """Get target configuration objects.

    Returns:
        List of Target configuration objects.
    """
    targets = []
    for target in listify_targets(self.config.get("targets", "")):
        if target.startswith("@"):
            group = self.world.groups.get(target[1:])
            if group:
                targets.extend(group.get_targets())
        else:
            targets.append(target)
    return targets

TargetConfig

Bases: BaseConfig

Target configuration.

Source code in penvm/src/client/penvm/client/config.py

class TargetConfig(BaseConfig):
    """Target configuration."""

    def __init__(self, *args, **kwargs):
        """Initialize."""
        super().__init__(*args, **kwargs)
        self.config["host"] = self.config.get("host", self.config.get("name"))

    def __repr__(self):
        return f"<TargetConfig name={self.config['name']} config={self.config}>"

__init__(*args, **kwargs)

Initialize.

Source code in penvm/src/client/penvm/client/config.py

def __init__(self, *args, **kwargs):
    """Initialize."""
    super().__init__(*args, **kwargs)
    self.config["host"] = self.config.get("host", self.config.get("name"))

TemplateConfig

Bases: BaseConfig

Template configuration.

Source code in penvm/src/client/penvm/client/config.py

class TemplateConfig(BaseConfig):
    """Template configuration."""

    def run(self):
        """Add target configurations based on template.

        Automatic substitution for `$target` is done for`host`,
        `machine-id` with target name.
        """
        targets = self.config.get("targets")
        for name in listify_targets(targets):
            config = copy.deepcopy(self.config)
            config["name"] = name
            if config.get("host") == "$target":
                config["host"] = name
            if config.get("machine-id") == "$target":
                config["machine-id"] = name
            config.pop("targets")
            self.world.add_target(name, config)

run()

Add target configurations based on template.

Automatic substitution for $target is done forhost, machine-id with target name.

Source code in penvm/src/client/penvm/client/config.py

def run(self):
    """Add target configurations based on template.

    Automatic substitution for `$target` is done for`host`,
    `machine-id` with target name.
    """
    targets = self.config.get("targets")
    for name in listify_targets(targets):
        config = copy.deepcopy(self.config)
        config["name"] = name
        if config.get("host") == "$target":
            config["host"] = name
        if config.get("machine-id") == "$target":
            config["machine-id"] = name
        config.pop("targets")
        self.world.add_target(name, config)

WorldConfig

World configuation.

Provide high-level methods for working with the world configuration file. This is normally used only by penvm.client.world.World.

Source code in penvm/src/client/penvm/client/config.py

class WorldConfig:
    """World configuation.

    Provide high-level methods for working with the world
    configuration file. This is normally used only by
    [penvm.client.world.World][].
    """

    def __init__(self):
        """Initialize."""
        self.groups = {}
        self.meta = {}
        self.networks = {}
        self.targets = {}
        self.templates = {}

    def __repr__(self):
        return f"<WorldConfig nnetworks={len(self.networks)}  ngroups={len(self.groups)} ntargets={len(self.targets)}>"

    def add_group(self, name: str, config: dict):
        """Add group.

        Args:
            name: Group name.
            config: Configuration.
        """
        self.groups[name] = GroupConfig(self, name, config)

    def add_meta(self, config: dict):
        """Add meta(data).

        Args:
            config: Configuration.
        """
        self.meta.update(config)

    def add_network(self, name: str, config: dict):
        """Add network.

        Args:
            name: Network name.
            config: Configuration.
        """
        self.networks[name] = NetworkConfig(self, name, config)

    def add_target(self, name: str, config: dict):
        """Add target.

        Args:
            name: Target name.
            config: Configuration.
        """
        self.targets[name] = TargetConfig(self, name, config)

    def add_template(self, name: str, config: dict):
        """Add template.

        Args:
            name: Template name.
            config: Configuration.
        """
        template = self.templates[name] = TemplateConfig(self, name, config)
        template.run()

    def clear(self):
        """Clear configuration."""
        self.networks = {}
        self.groups = {}
        self.targets = {}
        self.templates = {}
        self.meta = {}

    def load(self, path: str):
        """Load configuration from a file.

        Args:
            path: File path.
        """
        try:
            d = yaml.safe_load(open(path, "r"))
        except Exception as e:
            print(f"failed to load world file ({e})")
            raise

        self.load_config(d)

    def load_config(self, d: dict):
        """Load configuration from a dictionary.

        Args:
            d: Configuration.
        """
        self.add_meta(d.get("meta", {}))

        for name, config in d.get("templates", {}).items():
            self.add_template(name, config)

        for name, config in d.get("targets", {}).items():
            self.add_target(name, config)

        for name, config in d.get("groups", {}).items():
            self.add_group(name, config)

        for name, config in d.get("networks", {}).items():
            self.add_network(name, config)

    def get_group(self, name: str) -> "GroupConfig":
        """Get group.

        Args:
            name: Group name.

        Returns:
            Group confguration object.
        """
        return self.groups.get(name)

    def get_groups(self) -> List[str]:
        """Get group names.

        Returns:
            List of group names.
        """
        return self.groups.keys()

    def get_meta(self, name: str) -> Any:
        """Get meta(data).

        Args:
            name: Metadata item name.

        Returns:
            Metadata item value.
        """
        return self.meta.get(name)

    def get_network(self, name: str) -> "NetworkConfig":
        """Get network configuration.

        Args:
            name: Network name.

        Returns:
            Network configuration object.
        """
        return self.networks.get(name)

    def get_networks(self) -> List[str]:
        """Get network names.

        Returns:
            List of network names.
        """
        return self.networks.keys()

    def get_target(self, name: str) -> "TargetConfig":
        """Get target.

        Args:
            name (str): Target name.

        Returns:
            Target configuration object.
        """
        return self.targets.get(name)

    def get_targets(self) -> List[str]:
        """Get target names.

        Returns:
            List of target names.
        """
        return self.targets.keys()

__init__()

Initialize.

Source code in penvm/src/client/penvm/client/config.py

def __init__(self):
    """Initialize."""
    self.groups = {}
    self.meta = {}
    self.networks = {}
    self.targets = {}
    self.templates = {}

add_group(name, config)

Add group.

Parameters:

Name	Type	Description	Default
name	str	Group name.	required
config	dict	Configuration.	required

Source code in penvm/src/client/penvm/client/config.py

def add_group(self, name: str, config: dict):
    """Add group.

    Args:
        name: Group name.
        config: Configuration.
    """
    self.groups[name] = GroupConfig(self, name, config)

add_meta(config)

Add meta(data).

Parameters:

Name	Type	Description	Default
config	dict	Configuration.	required

Source code in penvm/src/client/penvm/client/config.py

def add_meta(self, config: dict):
    """Add meta(data).

    Args:
        config: Configuration.
    """
    self.meta.update(config)

add_network(name, config)

Add network.

Parameters:

Name	Type	Description	Default
name	str	Network name.	required
config	dict	Configuration.	required

Source code in penvm/src/client/penvm/client/config.py

def add_network(self, name: str, config: dict):
    """Add network.

    Args:
        name: Network name.
        config: Configuration.
    """
    self.networks[name] = NetworkConfig(self, name, config)

add_target(name, config)

Add target.

Parameters:

Name	Type	Description	Default
name	str	Target name.	required
config	dict	Configuration.	required

Source code in penvm/src/client/penvm/client/config.py

def add_target(self, name: str, config: dict):
    """Add target.

    Args:
        name: Target name.
        config: Configuration.
    """
    self.targets[name] = TargetConfig(self, name, config)

add_template(name, config)

Add template.

Parameters:

Name	Type	Description	Default
name	str	Template name.	required
config	dict	Configuration.	required

Source code in penvm/src/client/penvm/client/config.py

def add_template(self, name: str, config: dict):
    """Add template.

    Args:
        name: Template name.
        config: Configuration.
    """
    template = self.templates[name] = TemplateConfig(self, name, config)
    template.run()

clear()

Clear configuration.

Source code in penvm/src/client/penvm/client/config.py

def clear(self):
    """Clear configuration."""
    self.networks = {}
    self.groups = {}
    self.targets = {}
    self.templates = {}
    self.meta = {}

get_group(name)

Get group.

Parameters:

Name	Type	Description	Default
name	str	Group name.	required

Returns:

Type	Description
GroupConfig	Group confguration object.

Source code in penvm/src/client/penvm/client/config.py

def get_group(self, name: str) -> "GroupConfig":
    """Get group.

    Args:
        name: Group name.

    Returns:
        Group confguration object.
    """
    return self.groups.get(name)

get_groups()

Get group names.

Returns:

Type	Description
List[str]	List of group names.

Source code in penvm/src/client/penvm/client/config.py

def get_groups(self) -> List[str]:
    """Get group names.

    Returns:
        List of group names.
    """
    return self.groups.keys()

get_meta(name)

Get meta(data).

Parameters:

Name	Type	Description	Default
name	str	Metadata item name.	required

Returns:

Type	Description
Any	Metadata item value.

Source code in penvm/src/client/penvm/client/config.py

def get_meta(self, name: str) -> Any:
    """Get meta(data).

    Args:
        name: Metadata item name.

    Returns:
        Metadata item value.
    """
    return self.meta.get(name)

get_network(name)

Get network configuration.

Parameters:

Name	Type	Description	Default
name	str	Network name.	required

Returns:

Type	Description
NetworkConfig	Network configuration object.

Source code in penvm/src/client/penvm/client/config.py

def get_network(self, name: str) -> "NetworkConfig":
    """Get network configuration.

    Args:
        name: Network name.

    Returns:
        Network configuration object.
    """
    return self.networks.get(name)

get_networks()

Get network names.

Returns:

Type	Description
List[str]	List of network names.

Source code in penvm/src/client/penvm/client/config.py

def get_networks(self) -> List[str]:
    """Get network names.

    Returns:
        List of network names.
    """
    return self.networks.keys()

get_target(name)

Get target.

Parameters:

Name	Type	Description	Default
name	str	Target name.	required

Returns:

Type	Description
TargetConfig	Target configuration object.

Source code in penvm/src/client/penvm/client/config.py

def get_target(self, name: str) -> "TargetConfig":
    """Get target.

    Args:
        name (str): Target name.

    Returns:
        Target configuration object.
    """
    return self.targets.get(name)

get_targets()

Get target names.

Returns:

Type	Description
List[str]	List of target names.

Source code in penvm/src/client/penvm/client/config.py

def get_targets(self) -> List[str]:
    """Get target names.

    Returns:
        List of target names.
    """
    return self.targets.keys()

load(path)

Load configuration from a file.

Parameters:

Name	Type	Description	Default
path	str	File path.	required

Source code in penvm/src/client/penvm/client/config.py

def load(self, path: str):
    """Load configuration from a file.

    Args:
        path: File path.
    """
    try:
        d = yaml.safe_load(open(path, "r"))
    except Exception as e:
        print(f"failed to load world file ({e})")
        raise

    self.load_config(d)

load_config(d)

Load configuration from a dictionary.

Parameters:

Name	Type	Description	Default
d	dict	Configuration.	required

Source code in penvm/src/client/penvm/client/config.py

def load_config(self, d: dict):
    """Load configuration from a dictionary.

    Args:
        d: Configuration.
    """
    self.add_meta(d.get("meta", {}))

    for name, config in d.get("templates", {}).items():
        self.add_template(name, config)

    for name, config in d.get("targets", {}).items():
        self.add_target(name, config)

    for name, config in d.get("groups", {}).items():
        self.add_group(name, config)

    for name, config in d.get("networks", {}).items():
        self.add_network(name, config)

expand_range(s)

Expand numbered, comma-separated ranges.

Supported formats are:

• <start>-<end>
• <single>

Leading 0s set width of resulting numbers (e.g., 001 for a width of three digits).

Parameters:

Name	Type	Description	Default
s	str	Formatted string.	required

Returns:

Type	Description
List[str]	List of expanded ranges.

Source code in penvm/src/client/penvm/client/config.py

def expand_range(s: str) -> List[str]:
    """Expand numbered, comma-separated ranges.

    Supported formats are:

    * <start>-<end>
    * <single>

    Leading 0s set width of resulting numbers (e.g., 001 for a width
    of three digits).

    Args:
        s (str): Formatted string.

    Returns:
        List of expanded ranges.
    """
    l = []
    while s:
        if "[" in s:
            i = s.find("[")
            ii = s.find("]")
            l.append([s[:i]])
            ll = []
            segment = s[i + 1 : ii]
            chunks = segment.split(",")
            for chunk in chunks:
                chunk = chunk.strip()
                if "-" in chunk:
                    first, last = chunk.split("-")
                    sz = min(map(len, [first, last]))
                    first, last = int(first), int(last)
                    ll.extend(list(srange(first, last + 1, 1, sz)))
                else:
                    ll.append(chunk)
            l.append(ll)
            s = s[ii + 1 :]
        else:
            l.append([s])
            s = ""

    return ["".join(t) for t in itertools.product(*l)]

expand_value(v, d)

Expand/resolve value using dictionary.

Resolvable items are specified using the str.format.

Parameters:

Name	Type	Description	Default
v	str	Regular or format string.	required
d	dict	Dictionary used to resolve format string.	required

Returns:

Type	Description
str	Resolved "value".

Source code in penvm/src/client/penvm/client/config.py

def expand_value(v: str, d: dict) -> str:
    """Expand/resolve value using dictionary.

    Resolvable items are specified using the `str.format`.

    Args:
        v: Regular or format string.
        d: Dictionary used to resolve format string.

    Returns:
        Resolved "value".
    """
    if type(v) == str:
        return v.format(**d)
    return v

listify_targets(s)

Expand targets settings.

Parameters:

Name	Type	Description	Default
s	str	String of targets/target ranges.	required

Returns:

Type	Description
List[str]	List of expanded ranges.

Source code in penvm/src/client/penvm/client/config.py

def listify_targets(s: str) -> List[str]:
    """Expand targets settings.

    Args:
        s: String of targets/target ranges.

    Returns:
        List of expanded ranges.
    """
    l = []
    for name in s.split():
        l.extend(expand_range(name))
    return l

srange(start, end, step, sz)

Parameters:

Name	Type	Description	Default
start	int	Start value.	required
end	int	End value.	required
step	int	Step value.	required
sz	str	Field width.	required

Yields:

Type	Description
str	Range value.

Source code in penvm/src/client/penvm/client/config.py

def srange(start: int, end: int, step: int, sz: str):
    """
    Args:
        start: Start value.
        end: End value.
        step: Step value.
        sz: Field width.

    Yields:
        (str): Range value.
    """
    fmt = "%%0.%dd" % (sz,)
    for v in range(start, end, step):
        yield fmt % v

penvm.client.machine

Machine

Bases: BaseObject

Client-side representation of a server-side machine.

Source code in penvm/src/client/penvm/client/machine.py

class Machine(BaseObject):
    """Client-side representation of a server-side machine."""

    def __init__(
        self,
        host: str,
        port: int,
        sslprofile: Union[str, None] = None,
        machineid: Union[str, None] = None,
    ):
        """Initialize.

        Set up client-side machine representation.

        Args:
            host: Host address.
            port: Port.
            sslprofile: SSL profile for SSL context.
            machineid: Machine id. Generated if not provided.
        """
        try:
            super().__init__(machineid, logger)
            tlogger = self.logger.enter()

            self.sslprofile = sslprofile
            self.sslcontext = self.get_sslcontext(self.sslprofile)

            self.conn = ClientConnection(self, host, port, self.sslcontext)
            self.conn.connect()
            # time.sleep(0.5)
            self.conn.start()

            self.exit = False

            self.imq = RoutingQueue(put=self.imq_put)
            if self.conn:
                self.omq = RoutingQueue(put=self.omq_put)
            else:
                # for testing without a connection
                self.omq = MessageQueue()

            self.lock = threading.Lock()
            self.kernels = {}
            self.sessions = {}

            # standard kernels
            self.load_kernel("core", "penvm.kernels.core")
            self.load_kernel("default", "penvm.kernels.default")
        except Exception as e:
            self.logger.warning(f"EXCEPTION ({e})")
        finally:
            tlogger.exit()

    def __repr__(self):
        return f"<Machine id={self.oid} conn={self.conn} nsessions={len(self.sessions)}>"

    def get_debug_session(self, sessionid: str = None) -> "Session":
        """Get debug-specific session.

        Debug sessions (should) start with "-debug-". They are treated
        specially by machines: they are not subject to the debug mode.

        Args:
            sessionid: Session id (suffix) for a debug session.

        Returns:
            Session for debugging.
        """
        return self.get_session("""-debug-{sessionid or ""}""")

    def get_kernel_class(self, kernel_name: str) -> Type["KernelClient"]:
        """Get kernel client class.

        Args:
            kernel_name: Kernel name.

        Returns:
            Kernel client class for `kernel_name`.
        """
        return self.kernels.get(kernel_name)

    def get_machconnspec(self) -> MachineConnectionSpec:
        """Get `MachineConnectionSpec` object for this machine.

        Returns:
            `MachineConnectionSpec` of object for this machine.
        """
        return MachineConnectionSpec(machine=self)

    def get_machconnstr(self) -> str:
        """Get machine connection string for this machine.

        Returns:
            Machine connection string.
        """
        return str(self.get_machconnspec())

    def get_session(
        self, sessionid: Union[str, None] = None, kernelname: str = "default"
    ) -> "Session":
        """Get session.

        Args:
            sessionid: Session id. Generated if not provided.
            kernelname: Kernel name.

        Returns:
            New `Session` for `sessionid` and `kernelname`.
        """
        try:
            tlogger = self.logger.enter()
            tlogger.debug(f"getting session for sessionid={sessionid}")

            # lock
            self.lock.acquire()
            try:
                kernelcls = self.get_kernel_class(kernelname)
                if kernelcls == None:
                    tlogger.debug(f"kernel ({kernelname}) not found")
                    return

                session = self.sessions.get(sessionid) if sessionid != None else None
                if session == None:
                    session = Session(self, sessionid, kernelcls)
                    session = self.sessions.setdefault(session.oid, session)
            finally:
                self.lock.release()
            return session
        finally:
            tlogger.exit()

    def get_sslcontext(self, sslprofile: str) -> "SSLContext":
        """Load client-side SSLContext based on named ssl profile.

        Args:
            sslprofile (str): SSL profile name.

        Returns:
            `SSLContext` for ssl profile.
        """
        if sslprofile:
            try:
                import ssl

                sslcontext = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
                sslcontext.check_hostname = False
                sslcontext.verify_mode = ssl.CERT_NONE
                return sslcontext
            except Exception as e:
                logger.debug(f"ssl required but missing for ssl profile ({sslprofile})")
                raise Exception(f"ssl required but missing for ssl profile ({sslprofile})")

    def imq_put(self, msg: "Message"):
        """Put a message on the IMQ.

        Args:
            msg: Message object.
        """
        try:
            tlogger = self.logger.enter()

            sessionid = msg.header.get("session-id")

            tlogger.debug(f"imq_put sessionid={sessionid}")
            sess = self.sessions.get(sessionid)
            if sess:
                sess.imq.put(msg)
            else:
                # DROP!
                tlogger.debug("imp_put dropping message")
        finally:
            tlogger.exit()

    def list_kernels(self) -> List[str]:
        """Get list of kernel names.

        Returns:
            List of kernel names.
        """
        try:
            tlogger = self.logger.enter()
            return list(self.kernels.keys())
        finally:
            tlogger.exit()

    def load_assets(self, assets):
        """Load/copy assets to the machines.

        NIY.
        """
        try:
            tlogger = self.logger.enter()
        finally:
            tlogger.exit()

    def load_kernel(self, kernel_name: str, pkgname: str) -> "KernelClient":
        """Load/copy kernel to the machine.

        Kernel support is provided as:

        ```
            <pkg>/
              client.py
                KernelClient
              server.py
                Kernel
        ```

        Args:
            kernel_name: Kernel name.
            pkgname: Package name (as a string).

        Returns:
            Kernel client class.
        """
        try:
            tlogger = self.logger.enter()
            if kernel_name in self.kernels:
                # TODO: test? allow overwrite for now?
                pass

            try:
                mod = importlib.import_module(".client", pkgname)
                cls = getattr(mod, "KernelClient")
                self.kernels[kernel_name] = cls
            except Exception as e:
                pass

            # TODO: send kernel to machine (server side)
            return cls
        finally:
            tlogger.exit()

    def omq_put(self, msg: "Message"):
        """Put a message on the OMQ.

        Args:
            msg: Message to enqueue.
        """
        try:
            tlogger = self.logger.enter()
            self.conn.omq.put(msg)
        finally:
            tlogger.exit()

    def start(self):
        """Start machine."""
        try:
            tlogger = self.logger.enter()
            if 0 and self.conn:
                self.conn.connect()
                self.conn.start()
        finally:
            tlogger.exit()

    def stop(self):
        """Stop machine.

        Set `exit` attribute to signal machine should exit.
        """
        try:
            tlogger = self.logger.enter()
            self.exit = True
        finally:
            tlogger.exit()

__init__(host, port, sslprofile=None, machineid=None)

Initialize.

Set up client-side machine representation.

Parameters:

Name	Type	Description	Default
host	str	Host address.	required
port	int	Port.	required
sslprofile	Union[str, None]	SSL profile for SSL context.	None
machineid	Union[str, None]	Machine id. Generated if not provided.	None

Source code in penvm/src/client/penvm/client/machine.py

def __init__(
    self,
    host: str,
    port: int,
    sslprofile: Union[str, None] = None,
    machineid: Union[str, None] = None,
):
    """Initialize.

    Set up client-side machine representation.

    Args:
        host: Host address.
        port: Port.
        sslprofile: SSL profile for SSL context.
        machineid: Machine id. Generated if not provided.
    """
    try:
        super().__init__(machineid, logger)
        tlogger = self.logger.enter()

        self.sslprofile = sslprofile
        self.sslcontext = self.get_sslcontext(self.sslprofile)

        self.conn = ClientConnection(self, host, port, self.sslcontext)
        self.conn.connect()
        # time.sleep(0.5)
        self.conn.start()

        self.exit = False

        self.imq = RoutingQueue(put=self.imq_put)
        if self.conn:
            self.omq = RoutingQueue(put=self.omq_put)
        else:
            # for testing without a connection
            self.omq = MessageQueue()

        self.lock = threading.Lock()
        self.kernels = {}
        self.sessions = {}

        # standard kernels
        self.load_kernel("core", "penvm.kernels.core")
        self.load_kernel("default", "penvm.kernels.default")
    except Exception as e:
        self.logger.warning(f"EXCEPTION ({e})")
    finally:
        tlogger.exit()

get_debug_session(sessionid=None)

Get debug-specific session.

Debug sessions (should) start with "-debug-". They are treated specially by machines: they are not subject to the debug mode.

Parameters:

Name	Type	Description	Default
sessionid	str	Session id (suffix) for a debug session.	None

Returns:

Type	Description
Session	Session for debugging.

Source code in penvm/src/client/penvm/client/machine.py

def get_debug_session(self, sessionid: str = None) -> "Session":
    """Get debug-specific session.

    Debug sessions (should) start with "-debug-". They are treated
    specially by machines: they are not subject to the debug mode.

    Args:
        sessionid: Session id (suffix) for a debug session.

    Returns:
        Session for debugging.
    """
    return self.get_session("""-debug-{sessionid or ""}""")

get_kernel_class(kernel_name)

Get kernel client class.

Parameters:

Name	Type	Description	Default
kernel_name	str	Kernel name.	required

Returns:

Type	Description
Type[KernelClient]	Kernel client class for kernel_name.

Source code in penvm/src/client/penvm/client/machine.py

def get_kernel_class(self, kernel_name: str) -> Type["KernelClient"]:
    """Get kernel client class.

    Args:
        kernel_name: Kernel name.

    Returns:
        Kernel client class for `kernel_name`.
    """
    return self.kernels.get(kernel_name)

get_machconnspec()

Get MachineConnectionSpec object for this machine.

Returns:

Type	Description
MachineConnectionSpec	MachineConnectionSpec of object for this machine.

Source code in penvm/src/client/penvm/client/machine.py

def get_machconnspec(self) -> MachineConnectionSpec:
    """Get `MachineConnectionSpec` object for this machine.

    Returns:
        `MachineConnectionSpec` of object for this machine.
    """
    return MachineConnectionSpec(machine=self)

get_machconnstr()

Get machine connection string for this machine.

Returns:

Type	Description
str	Machine connection string.

Source code in penvm/src/client/penvm/client/machine.py

def get_machconnstr(self) -> str:
    """Get machine connection string for this machine.

    Returns:
        Machine connection string.
    """
    return str(self.get_machconnspec())

get_session(sessionid=None, kernelname='default')

Get session.

Parameters:

Name	Type	Description	Default
sessionid	Union[str, None]	Session id. Generated if not provided.	None
kernelname	str	Kernel name.	'default'

Returns:

Type	Description
Session	New Session for sessionid and kernelname.

Source code in penvm/src/client/penvm/client/machine.py

def get_session(
    self, sessionid: Union[str, None] = None, kernelname: str = "default"
) -> "Session":
    """Get session.

    Args:
        sessionid: Session id. Generated if not provided.
        kernelname: Kernel name.

    Returns:
        New `Session` for `sessionid` and `kernelname`.
    """
    try:
        tlogger = self.logger.enter()
        tlogger.debug(f"getting session for sessionid={sessionid}")

        # lock
        self.lock.acquire()
        try:
            kernelcls = self.get_kernel_class(kernelname)
            if kernelcls == None:
                tlogger.debug(f"kernel ({kernelname}) not found")
                return

            session = self.sessions.get(sessionid) if sessionid != None else None
            if session == None:
                session = Session(self, sessionid, kernelcls)
                session = self.sessions.setdefault(session.oid, session)
        finally:
            self.lock.release()
        return session
    finally:
        tlogger.exit()

get_sslcontext(sslprofile)

Load client-side SSLContext based on named ssl profile.

Parameters:

Name	Type	Description	Default
sslprofile	str	SSL profile name.	required

Returns:

Type	Description
SSLContext	SSLContext for ssl profile.

Source code in penvm/src/client/penvm/client/machine.py

def get_sslcontext(self, sslprofile: str) -> "SSLContext":
    """Load client-side SSLContext based on named ssl profile.

    Args:
        sslprofile (str): SSL profile name.

    Returns:
        `SSLContext` for ssl profile.
    """
    if sslprofile:
        try:
            import ssl

            sslcontext = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
            sslcontext.check_hostname = False
            sslcontext.verify_mode = ssl.CERT_NONE
            return sslcontext
        except Exception as e:
            logger.debug(f"ssl required but missing for ssl profile ({sslprofile})")
            raise Exception(f"ssl required but missing for ssl profile ({sslprofile})")

imq_put(msg)

Put a message on the IMQ.

Parameters:

Name	Type	Description	Default
msg	Message	Message object.	required

Source code in penvm/src/client/penvm/client/machine.py

def imq_put(self, msg: "Message"):
    """Put a message on the IMQ.

    Args:
        msg: Message object.
    """
    try:
        tlogger = self.logger.enter()

        sessionid = msg.header.get("session-id")

        tlogger.debug(f"imq_put sessionid={sessionid}")
        sess = self.sessions.get(sessionid)
        if sess:
            sess.imq.put(msg)
        else:
            # DROP!
            tlogger.debug("imp_put dropping message")
    finally:
        tlogger.exit()

list_kernels()

Get list of kernel names.

Returns:

Type	Description
List[str]	List of kernel names.

Source code in penvm/src/client/penvm/client/machine.py

def list_kernels(self) -> List[str]:
    """Get list of kernel names.

    Returns:
        List of kernel names.
    """
    try:
        tlogger = self.logger.enter()
        return list(self.kernels.keys())
    finally:
        tlogger.exit()

load_assets(assets)

Load/copy assets to the machines.

NIY.

Source code in penvm/src/client/penvm/client/machine.py

def load_assets(self, assets):
    """Load/copy assets to the machines.

    NIY.
    """
    try:
        tlogger = self.logger.enter()
    finally:
        tlogger.exit()

load_kernel(kernel_name, pkgname)

Load/copy kernel to the machine.

Kernel support is provided as:

    <pkg>/
      client.py
        KernelClient
      server.py
        Kernel

Parameters:

Name	Type	Description	Default
kernel_name	str	Kernel name.	required
pkgname	str	Package name (as a string).	required

Returns:

Type	Description
KernelClient	Kernel client class.

Source code in penvm/src/client/penvm/client/machine.py

def load_kernel(self, kernel_name: str, pkgname: str) -> "KernelClient":
    """Load/copy kernel to the machine.

    Kernel support is provided as:

    ```
        <pkg>/
          client.py
            KernelClient
          server.py
            Kernel
    ```

    Args:
        kernel_name: Kernel name.
        pkgname: Package name (as a string).

    Returns:
        Kernel client class.
    """
    try:
        tlogger = self.logger.enter()
        if kernel_name in self.kernels:
            # TODO: test? allow overwrite for now?
            pass

        try:
            mod = importlib.import_module(".client", pkgname)
            cls = getattr(mod, "KernelClient")
            self.kernels[kernel_name] = cls
        except Exception as e:
            pass

        # TODO: send kernel to machine (server side)
        return cls
    finally:
        tlogger.exit()

omq_put(msg)

Put a message on the OMQ.

Parameters:

Name	Type	Description	Default
msg	Message	Message to enqueue.	required

Source code in penvm/src/client/penvm/client/machine.py

def omq_put(self, msg: "Message"):
    """Put a message on the OMQ.

    Args:
        msg: Message to enqueue.
    """
    try:
        tlogger = self.logger.enter()
        self.conn.omq.put(msg)
    finally:
        tlogger.exit()

start()

Start machine.

Source code in penvm/src/client/penvm/client/machine.py

def start(self):
    """Start machine."""
    try:
        tlogger = self.logger.enter()
        if 0 and self.conn:
            self.conn.connect()
            self.conn.start()
    finally:
        tlogger.exit()

stop()

Stop machine.

Set exit attribute to signal machine should exit.

Source code in penvm/src/client/penvm/client/machine.py

def stop(self):
    """Stop machine.

    Set `exit` attribute to signal machine should exit.
    """
    try:
        tlogger = self.logger.enter()
        self.exit = True
    finally:
        tlogger.exit()

penvm.client.session

A client-side session is used to interact with a machine. Each session has its own unique session id, specified on the client side. Any number of concurrent sessions are supported, subject to resource limits.

Each client-side session has its own incoming and outgoing message queues (not the same as those on the server side but related). All communication is mediated with these message queues with forwarding to and from handle automatically (see penvm.lib.connection).

Convenience methods are provided for interacting at a high level with the message queues (e.g., Session.get_response()).

Each client-side session is set up with access to a selected, session-specific, client-side kernel interface. Although not absolutely required, this interface is highly recommended to provide a function-style interface rather than having to build a Request message.

See penvm.server.session for details.

Session

Bases: BaseObject

Client-side session object.

Provides unique sessionid and access to client-side machine services.

Low-level methods return the request object. This allows for tracking and followup of a session.

High-level methods return None.

Source code in penvm/src/client/penvm/client/session.py

class Session(BaseObject):
    """Client-side session object.

    Provides unique sessionid and access to client-side machine services.

    Low-level methods return the request object. This allows for tracking and
    followup of a session.

    High-level methods return None.
    """

    def __init__(
        self,
        machine: "Machine",
        sessionid: Union[str, None] = None,
        kernelcls: Type["KernelClient"] = None,
    ):
        """Initialize.

        Args:
            machine: Machine owning this session.
            sessionid: Session id.
            kernelcls: `KernelClient` class.
        """
        try:
            super().__init__(sessionid, logger)
            tlogger = self.logger.enter()

            self.machine = machine
            self.imq = MessageQueue()
            self.kernel = kernelcls(self)
        except Exception as e:
            self.logger.warning(f"EXCEPTION ({e})")
        finally:
            tlogger.exit()

    def __repr__(self):
        return f"<Session id={self.oid} machine={self.machine}>"

    def get_response(self) -> "Message":
        """Pull (from remote) and pop from local and return.

        Returns:
            Response message.
        """
        self.pull_response()
        return self.pop_response()

    def new_request(self, op: str, d: Union[dict, None] = None) -> "Message":
        """Return session-specific Request object with `op`.

        Args:
            op: Operation name.
            d: Request payload settings.

        Returns:
            Created request.
        """
        req = Request()
        req.header["session-id"] = self.oid
        req.payload["op"] = op
        if d != None:
            req.payload.update(d)
        return req

    def newput_request(self, op: str, d: Union[dict, None] = None) -> "Message":
        """Create request and *put it out* to be sent.

        Args:
            op: Operation name.
            d: Request payload settings.

        Returns:
            Created request.
        """
        req = self.new_request(op, d)
        return self.put_request(req)

    def pop_response(self) -> "Message":
        """Pop response from local and return.

        Returns:
            Response message.
        """
        return self.imq.pop()

    def pull_response(self, sessionid: Union[str, None] = None):
        """Pull (pop from remote to local) response.

        Args:
            sessionid: Session id.
        """
        self.kernel.session_pop_omq(sessionid)

    def put_request(self, req: "Message") -> "Message":
        """Put the request on the machine OMQ. Return it, also.

        Args:
            req: Request message.

        Returns:
            The Request message, itself.
        """
        self.machine.omq.put(req)
        return req

__init__(machine, sessionid=None, kernelcls=None)

Initialize.

Parameters:

Name	Type	Description	Default
machine	Machine	Machine owning this session.	required
sessionid	Union[str, None]	Session id.	None
kernelcls	Type[KernelClient]	KernelClient class.	None

Source code in penvm/src/client/penvm/client/session.py

def __init__(
    self,
    machine: "Machine",
    sessionid: Union[str, None] = None,
    kernelcls: Type["KernelClient"] = None,
):
    """Initialize.

    Args:
        machine: Machine owning this session.
        sessionid: Session id.
        kernelcls: `KernelClient` class.
    """
    try:
        super().__init__(sessionid, logger)
        tlogger = self.logger.enter()

        self.machine = machine
        self.imq = MessageQueue()
        self.kernel = kernelcls(self)
    except Exception as e:
        self.logger.warning(f"EXCEPTION ({e})")
    finally:
        tlogger.exit()

get_response()

Pull (from remote) and pop from local and return.

Returns:

Type	Description
Message	Response message.

Source code in penvm/src/client/penvm/client/session.py

def get_response(self) -> "Message":
    """Pull (from remote) and pop from local and return.

    Returns:
        Response message.
    """
    self.pull_response()
    return self.pop_response()

new_request(op, d=None)

Return session-specific Request object with op.

Parameters:

Name	Type	Description	Default
op	str	Operation name.	required
d	Union[dict, None]	Request payload settings.	None

Returns:

Type	Description
Message	Created request.

Source code in penvm/src/client/penvm/client/session.py

def new_request(self, op: str, d: Union[dict, None] = None) -> "Message":
    """Return session-specific Request object with `op`.

    Args:
        op: Operation name.
        d: Request payload settings.

    Returns:
        Created request.
    """
    req = Request()
    req.header["session-id"] = self.oid
    req.payload["op"] = op
    if d != None:
        req.payload.update(d)
    return req

newput_request(op, d=None)

Create request and put it out to be sent.

Parameters:

Name	Type	Description	Default
op	str	Operation name.	required
d	Union[dict, None]	Request payload settings.	None

Returns:

Type	Description
Message	Created request.

Source code in penvm/src/client/penvm/client/session.py

def newput_request(self, op: str, d: Union[dict, None] = None) -> "Message":
    """Create request and *put it out* to be sent.

    Args:
        op: Operation name.
        d: Request payload settings.

    Returns:
        Created request.
    """
    req = self.new_request(op, d)
    return self.put_request(req)

pop_response()

Pop response from local and return.

Returns:

Type	Description
Message	Response message.

Source code in penvm/src/client/penvm/client/session.py

def pop_response(self) -> "Message":
    """Pop response from local and return.

    Returns:
        Response message.
    """
    return self.imq.pop()

pull_response(sessionid=None)

Pull (pop from remote to local) response.

Parameters:

Name	Type	Description	Default
sessionid	Union[str, None]	Session id.	None

Source code in penvm/src/client/penvm/client/session.py

def pull_response(self, sessionid: Union[str, None] = None):
    """Pull (pop from remote to local) response.

    Args:
        sessionid: Session id.
    """
    self.kernel.session_pop_omq(sessionid)

put_request(req)

Put the request on the machine OMQ. Return it, also.

Parameters:

Name	Type	Description	Default
req	Message	Request message.	required

Returns:

Type	Description
Message	The Request message, itself.

Source code in penvm/src/client/penvm/client/session.py

def put_request(self, req: "Message") -> "Message":
    """Put the request on the machine OMQ. Return it, also.

    Args:
        req: Request message.

    Returns:
        The Request message, itself.
    """
    self.machine.omq.put(req)
    return req

penvm.client.world

The World object is foundational for defining the set of networks and machines available to an application. It can be set up in various ways: configuration, configuration file, network string, environment variables. All of these provide flexibility for many use cases.

An application is associated with a world configuration explicitly (specific configuration or named configuration file) or implicitly (environment variable or related .penvm file).

Each running machine instance is defined by a MachineConnectionSpec which can be represented as a Machine Connection string. A machine connection consists of:

• machine id (unique)
• ssl profile (name)
• hostname/address
• port

A network string consists of one or more machine connection strings.

Network

Bases: BaseObject

Network.

Encapsulates references to targets and booted machine instances.

Source code in penvm/src/client/penvm/client/world.py

class Network(BaseObject):
    """Network.

    Encapsulates references to targets and booted machine instances.
    """

    def __init__(self, name: str, targets: List["Target"], config: "NetworkConfig"):
        """Initialize.

        Args:
            name: Network name.
            targets: List of network targets.
            config: Network configuration.
        """
        try:
            super().__init__(name, logger)
            tlogger = self.logger.enter()

            self.name = name
            self.targets = targets
            self.config = config

            # augment from net config
            overrides = self.config.get("overrides")
            if overrides:
                for target in self.targets:
                    target.update(overrides)

            self.machines = {}
        finally:
            tlogger.exit()

    def __len__(self):
        return len(self.machines)

    def __str__(self):
        return " ".join(self.get_machconnstrs())

    def boot(
        self,
        concurrency: Union[int, None] = None,
    ):
        """Boot (machines in) network.

        Boots machine instances using target information.

        Boot concurrency ("boot-concurrency" configuration
        setting) is: 0: number of targets, 1: serial, > 1
        concurrent. Final minimum is 1. Default is 1.

        Args:
            concurrency: Boot concurrency. Overrides configuration if
                provided.
        """
        # TODO: run threaded to improve startup/setup times and smooth
        # out wait/delay/connect times
        try:
            tlogger = self.logger.enter()

            if self.machines:
                # already booted
                return

            if concurrency == None:
                concurrency = self.config.get("boot-concurrency", 1)
                if concurrency == 0:
                    concurrency = len(self.targets)
            concurrency = max(1, concurrency)

            # print(f"{concurrency=}")
            if concurrency == 1:
                # sequential boot
                for target in self.targets:
                    self.machines[target.name] = target.boot()
            else:
                # TODO: figure out why terminal gets messed up
                # concurrent boot
                def _boot(target):
                    machine = target.boot()
                    lock.acquire()
                    self.machines[target.name] = machine
                    lock.release()
                    self.logger.debug(f"booted target ({target.name}) machine ({machine.oid})")
                    # print(f"{machine=}")
                    return machine

                lock = threading.Lock()

                with concurrent.futures.ThreadPoolExecutor(max_workers=concurrency) as executor:
                    # with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
                    results = {executor.submit(_boot, target) for target in self.targets}
                    for future in concurrent.futures.as_completed(results):
                        # print(f"{future.result()=}")
                        pass

            # TODO: launch penvm-server
            # TODO: capture host/addr and port
            # TODO: set up Machines for each
        except Exception as e:
            self.logger.warning(f"EXCEPTION ({e})")
        finally:
            tlogger.exit()

    def get_machconnspecs(self) -> List["MachineConnectionSpec"]:
        """Get `MachineConnectionSpec`s for machines.

        Returns:
            List of `MachineConnectionSpec`s.
        """
        return [m.get_machconnspec() for m in self.machines.values()]

    def get_machconnstrs(self) -> List[str]:
        """Get machine connection strings for machines.

        Returns:
            List of machine connection strings.
        """
        return [m.get_machconnstr() for m in self.machines.values()]

    def get_machine(self, target_name: str) -> "Machine":
        """Get machine.

        Args:
            target_name (str): Target name.

        Returns:
            Machine for named target.
        """
        return self.machines.get(target_name)

    def get_machines(self) -> List["Machine"]:
        """Get list of machines for network.

        Returns:
            List of `Machine`s.
        """
        return list(self.machines.values())

    def get_machine_names(self) -> List[str]:
        """Get list of machine names for network.

        Returns:
            List of machine names.
        """
        return list(self.machines.keys())

    def get_targets(self) -> List["Target"]:
        """Get targets for network.

        Returns:
            List of Targets.
        """
        return self.targets

    def get_target_names(self) -> List[str]:
        """Get target names for network.

        Returns:
            List of target names.
        """
        return [target.name for target in self.targets]

    def is_alive(self) -> bool:
        """Indicate if network is alive, or not.

        Returns:
            `True` or `False`.
        """
        return self.machines and True or False

    def shutdown(self):
        """Shut down network.

        Request for machines to be shutdown. Clear local tracking
        of machines by network.
        """
        self.logger.debug("shutdown")
        for k, machine in self.machines.items()[:]:
            s = machine.get_session()
            s.kernel.machine_shutdown()
            self.machines.pop(k, None)

__init__(name, targets, config)

Initialize.

Parameters:

Name	Type	Description	Default
name	str	Network name.	required
targets	List[Target]	List of network targets.	required
config	NetworkConfig	Network configuration.	required

Source code in penvm/src/client/penvm/client/world.py

def __init__(self, name: str, targets: List["Target"], config: "NetworkConfig"):
    """Initialize.

    Args:
        name: Network name.
        targets: List of network targets.
        config: Network configuration.
    """
    try:
        super().__init__(name, logger)
        tlogger = self.logger.enter()

        self.name = name
        self.targets = targets
        self.config = config

        # augment from net config
        overrides = self.config.get("overrides")
        if overrides:
            for target in self.targets:
                target.update(overrides)

        self.machines = {}
    finally:
        tlogger.exit()

boot(concurrency=None)

Boot (machines in) network.

Boots machine instances using target information.

Boot concurrency ("boot-concurrency" configuration setting) is: 0: number of targets, 1: serial, > 1 concurrent. Final minimum is 1. Default is 1.

Parameters:

Name	Type	Description	Default
concurrency	Union[int, None]	Boot concurrency. Overrides configuration if provided.	None

Source code in penvm/src/client/penvm/client/world.py

def boot(
    self,
    concurrency: Union[int, None] = None,
):
    """Boot (machines in) network.

    Boots machine instances using target information.

    Boot concurrency ("boot-concurrency" configuration
    setting) is: 0: number of targets, 1: serial, > 1
    concurrent. Final minimum is 1. Default is 1.

    Args:
        concurrency: Boot concurrency. Overrides configuration if
            provided.
    """
    # TODO: run threaded to improve startup/setup times and smooth
    # out wait/delay/connect times
    try:
        tlogger = self.logger.enter()

        if self.machines:
            # already booted
            return

        if concurrency == None:
            concurrency = self.config.get("boot-concurrency", 1)
            if concurrency == 0:
                concurrency = len(self.targets)
        concurrency = max(1, concurrency)

        # print(f"{concurrency=}")
        if concurrency == 1:
            # sequential boot
            for target in self.targets:
                self.machines[target.name] = target.boot()
        else:
            # TODO: figure out why terminal gets messed up
            # concurrent boot
            def _boot(target):
                machine = target.boot()
                lock.acquire()
                self.machines[target.name] = machine
                lock.release()
                self.logger.debug(f"booted target ({target.name}) machine ({machine.oid})")
                # print(f"{machine=}")
                return machine

            lock = threading.Lock()

            with concurrent.futures.ThreadPoolExecutor(max_workers=concurrency) as executor:
                # with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
                results = {executor.submit(_boot, target) for target in self.targets}
                for future in concurrent.futures.as_completed(results):
                    # print(f"{future.result()=}")
                    pass

        # TODO: launch penvm-server
        # TODO: capture host/addr and port
        # TODO: set up Machines for each
    except Exception as e:
        self.logger.warning(f"EXCEPTION ({e})")
    finally:
        tlogger.exit()

get_machconnspecs()

Get MachineConnectionSpecs for machines.

Returns:

Type	Description
List[MachineConnectionSpec]	List of MachineConnectionSpecs.

Source code in penvm/src/client/penvm/client/world.py

def get_machconnspecs(self) -> List["MachineConnectionSpec"]:
    """Get `MachineConnectionSpec`s for machines.

    Returns:
        List of `MachineConnectionSpec`s.
    """
    return [m.get_machconnspec() for m in self.machines.values()]

get_machconnstrs()

Get machine connection strings for machines.

Returns:

Type	Description
List[str]	List of machine connection strings.

Source code in penvm/src/client/penvm/client/world.py

def get_machconnstrs(self) -> List[str]:
    """Get machine connection strings for machines.

    Returns:
        List of machine connection strings.
    """
    return [m.get_machconnstr() for m in self.machines.values()]

get_machine(target_name)

Get machine.

Parameters:

Name	Type	Description	Default
target_name	str	Target name.	required

Returns:

Type	Description
Machine	Machine for named target.

Source code in penvm/src/client/penvm/client/world.py

def get_machine(self, target_name: str) -> "Machine":
    """Get machine.

    Args:
        target_name (str): Target name.

    Returns:
        Machine for named target.
    """
    return self.machines.get(target_name)

get_machine_names()

Get list of machine names for network.

Returns:

Type	Description
List[str]	List of machine names.

Source code in penvm/src/client/penvm/client/world.py

def get_machine_names(self) -> List[str]:
    """Get list of machine names for network.

    Returns:
        List of machine names.
    """
    return list(self.machines.keys())

get_machines()

Get list of machines for network.

Returns:

Type	Description
List[Machine]	List of Machines.

Source code in penvm/src/client/penvm/client/world.py

def get_machines(self) -> List["Machine"]:
    """Get list of machines for network.

    Returns:
        List of `Machine`s.
    """
    return list(self.machines.values())

get_target_names()

Get target names for network.

Returns:

Type	Description
List[str]	List of target names.

Source code in penvm/src/client/penvm/client/world.py

def get_target_names(self) -> List[str]:
    """Get target names for network.

    Returns:
        List of target names.
    """
    return [target.name for target in self.targets]

get_targets()

Get targets for network.

Returns:

Type	Description
List[Target]	List of Targets.

Source code in penvm/src/client/penvm/client/world.py

def get_targets(self) -> List["Target"]:
    """Get targets for network.

    Returns:
        List of Targets.
    """
    return self.targets

is_alive()

Indicate if network is alive, or not.

Returns:

Type	Description
bool	True or False.

Source code in penvm/src/client/penvm/client/world.py

def is_alive(self) -> bool:
    """Indicate if network is alive, or not.

    Returns:
        `True` or `False`.
    """
    return self.machines and True or False

shutdown()

Shut down network.

Request for machines to be shutdown. Clear local tracking of machines by network.

Source code in penvm/src/client/penvm/client/world.py

def shutdown(self):
    """Shut down network.

    Request for machines to be shutdown. Clear local tracking
    of machines by network.
    """
    self.logger.debug("shutdown")
    for k, machine in self.machines.items()[:]:
        s = machine.get_session()
        s.kernel.machine_shutdown()
        self.machines.pop(k, None)

Target

Bases: BaseObject

Target.

Source code in penvm/src/client/penvm/client/world.py

class Target(BaseObject):
    """Target."""

    def __init__(self, name: str, config: Union["TargetConfiguration", None] = None):
        """Initialize.

        Args:
            name: Target name.
            config: Target configuration.
        """
        try:
            super().__init__(name, logger)
            tlogger = self.logger.enter()

            self.name = name
            self.config = config or {}
        except Exception as e:
            self.logger.warning(f"EXCEPTION ({e})")
        finally:
            tlogger.exit()

    def __repr__(self):
        # url = self.config.get("url") if self.config else ""
        scheme = self.config.get("scheme")
        host = self.config.get("host")
        port = self.config.get("port")
        user = self.config.get("user")
        return f"<Target name={self.name} scheme={scheme} host={host} port={port} user={user}>"

    def boot(self):
        """Boot machine on target.

        Spawns `penvm-server` processes for each target.
        """
        try:
            tlogger = self.logger.enter()

            scheme = self.config.get("scheme")
            host = self.config.get("host")
            port = self.config.get("port")
            user = self.config.get("user")
            sslprofile = self.config.get("ssl-profile")

            # assign unique machine id
            machineid = self.config.get("machine-id", get_uuid())
            serverpath = f".penvm/releases/{get_version_string()}/penvm-server"

            t0 = time.time()
            if scheme == "auto":
                pass
            else:
                if scheme == "ssh":
                    spargs = [
                        "ssh",
                        host,
                        # "-T",
                        "-tt",
                        "-x",
                    ]
                    if user:
                        spargs.extend(["-l", user])
                    if port:
                        spargs.extend(["-p", port])

                elif scheme == "local":
                    serverpath = os.path.expanduser(f"~/{serverpath}")
                    spargs = []
                else:
                    return

                spargs.extend(
                    [
                        "python3",
                        serverpath,
                        "--machineid",
                        machineid,
                        "--announce",
                        "--background",
                        "--firstwait",
                        # "15",
                        "5",
                    ]
                )

                if sslprofile:
                    spargs.extend(["--ssl-profile", sslprofile])

                spargs.append(host)

                tlogger.debug("booting server ...")
                tlogger.debug(f"boot args={spargs}")
                # print(f"boot args={spargs}")
                cp = subprocess.run(
                    spargs,
                    stdin=subprocess.DEVNULL,
                    capture_output=True,
                    text=True,
                    cwd="/",
                    timeout=5,
                )

                if cp.returncode != 0:
                    tlogger.debug(f"boot failed returncode={cp.returncode} stderr={cp.stderr}")

                t1 = time.time()
                # tlogger.debug(f"machine booted elapsed ({t1-t0})")

            # set up Machine info
            try:
                # print(f"{scheme=}")
                # print(f"{self.config=}")
                if scheme == "auto":
                    mcs = MachineConnectionSpec(config=self.config)
                else:
                    # print(f"{cp.stdout=}")
                    # print(f"{cp.stderr=}")
                    if not cp.stdout.startswith("announce::"):
                        raise Exception(
                            f"could not get penvm-server information ({cp.stderr}). (is penvm-server deployed?)"
                        )

                    # trim announcement
                    s = cp.stdout[10:]
                    mcs = MachineConnectionSpec(machconnstr=s)

                for _ in range(10):
                    # TODO: be smart about `sleep`/waiting for remote readiness
                    if scheme != "auto":
                        time.sleep(0.05)
                    tlogger.debug(f"machine announcement received machineconnectionstr={mcs}")

                    tlogger.debug("creating Machine ...")
                    mach = Machine(
                        mcs.host,
                        mcs.port,
                        mcs.sslprofile,
                        mcs.machid,
                    )
                    if mach:
                        mach.start()
                        break

                t1 = time.time()
                tlogger.debug(f"machine created elapsed ({t1-t0})")

                if mach:
                    # set up sessions
                    sessions = self.config.get("sessions")
                    if sessions:
                        sess = mach.get_session("_")
                        for sessionid, session in sessions.items():
                            kernelname = session.get("kernel")
                            maxthreads = session.get("max-threads")

                            if kernelname != None:
                                tlogger.debug(
                                    f"session ({sessionid}) setting kernel ({kernelname})"
                                )
                                sess.kernel.session_use_kernel(kernelname, sessionid=sessionid)
                            if maxthreads != None:
                                tlogger.debug(
                                    f"session ({sessionid}) setting max threads ({maxthreads})"
                                )
                                sess.kernel.session_set_max_threads(
                                    maxthreads, sessionid=sessionid
                                )

                    # load assets
                    pass

                return mach
            except Exception as e:
                raise
        finally:
            tlogger.exit()

    def update(self, config: dict):
        """Update configuration.

        Args:
            config: Configuration.
        """
        self.config.update(config)

__init__(name, config=None)

Initialize.

Parameters:

Name	Type	Description	Default
name	str	Target name.	required
config	Union[TargetConfiguration, None]	Target configuration.	None

Source code in penvm/src/client/penvm/client/world.py

def __init__(self, name: str, config: Union["TargetConfiguration", None] = None):
    """Initialize.

    Args:
        name: Target name.
        config: Target configuration.
    """
    try:
        super().__init__(name, logger)
        tlogger = self.logger.enter()

        self.name = name
        self.config = config or {}
    except Exception as e:
        self.logger.warning(f"EXCEPTION ({e})")
    finally:
        tlogger.exit()

boot()

Boot machine on target.

Spawns penvm-server processes for each target.

Source code in penvm/src/client/penvm/client/world.py

def boot(self):
    """Boot machine on target.

    Spawns `penvm-server` processes for each target.
    """
    try:
        tlogger = self.logger.enter()

        scheme = self.config.get("scheme")
        host = self.config.get("host")
        port = self.config.get("port")
        user = self.config.get("user")
        sslprofile = self.config.get("ssl-profile")

        # assign unique machine id
        machineid = self.config.get("machine-id", get_uuid())
        serverpath = f".penvm/releases/{get_version_string()}/penvm-server"

        t0 = time.time()
        if scheme == "auto":
            pass
        else:
            if scheme == "ssh":
                spargs = [
                    "ssh",
                    host,
                    # "-T",
                    "-tt",
                    "-x",
                ]
                if user:
                    spargs.extend(["-l", user])
                if port:
                    spargs.extend(["-p", port])

            elif scheme == "local":
                serverpath = os.path.expanduser(f"~/{serverpath}")
                spargs = []
            else:
                return

            spargs.extend(
                [
                    "python3",
                    serverpath,
                    "--machineid",
                    machineid,
                    "--announce",
                    "--background",
                    "--firstwait",
                    # "15",
                    "5",
                ]
            )

            if sslprofile:
                spargs.extend(["--ssl-profile", sslprofile])

            spargs.append(host)

            tlogger.debug("booting server ...")
            tlogger.debug(f"boot args={spargs}")
            # print(f"boot args={spargs}")
            cp = subprocess.run(
                spargs,
                stdin=subprocess.DEVNULL,
                capture_output=True,
                text=True,
                cwd="/",
                timeout=5,
            )

            if cp.returncode != 0:
                tlogger.debug(f"boot failed returncode={cp.returncode} stderr={cp.stderr}")

            t1 = time.time()
            # tlogger.debug(f"machine booted elapsed ({t1-t0})")

        # set up Machine info
        try:
            # print(f"{scheme=}")
            # print(f"{self.config=}")
            if scheme == "auto":
                mcs = MachineConnectionSpec(config=self.config)
            else:
                # print(f"{cp.stdout=}")
                # print(f"{cp.stderr=}")
                if not cp.stdout.startswith("announce::"):
                    raise Exception(
                        f"could not get penvm-server information ({cp.stderr}). (is penvm-server deployed?)"
                    )

                # trim announcement
                s = cp.stdout[10:]
                mcs = MachineConnectionSpec(machconnstr=s)

            for _ in range(10):
                # TODO: be smart about `sleep`/waiting for remote readiness
                if scheme != "auto":
                    time.sleep(0.05)
                tlogger.debug(f"machine announcement received machineconnectionstr={mcs}")

                tlogger.debug("creating Machine ...")
                mach = Machine(
                    mcs.host,
                    mcs.port,
                    mcs.sslprofile,
                    mcs.machid,
                )
                if mach:
                    mach.start()
                    break

            t1 = time.time()
            tlogger.debug(f"machine created elapsed ({t1-t0})")

            if mach:
                # set up sessions
                sessions = self.config.get("sessions")
                if sessions:
                    sess = mach.get_session("_")
                    for sessionid, session in sessions.items():
                        kernelname = session.get("kernel")
                        maxthreads = session.get("max-threads")

                        if kernelname != None:
                            tlogger.debug(
                                f"session ({sessionid}) setting kernel ({kernelname})"
                            )
                            sess.kernel.session_use_kernel(kernelname, sessionid=sessionid)
                        if maxthreads != None:
                            tlogger.debug(
                                f"session ({sessionid}) setting max threads ({maxthreads})"
                            )
                            sess.kernel.session_set_max_threads(
                                maxthreads, sessionid=sessionid
                            )

                # load assets
                pass

            return mach
        except Exception as e:
            raise
    finally:
        tlogger.exit()

update(config)

Update configuration.

Parameters:

Name	Type	Description	Default
config	dict	Configuration.	required

Source code in penvm/src/client/penvm/client/world.py

def update(self, config: dict):
    """Update configuration.

    Args:
        config: Configuration.
    """
    self.config.update(config)

World

Bases: BaseObject

World of networks and hosts.

Configuration is taken from:

1. config keyword argument
2. networkstr keyword argument
3. filename keyword argument
4. PENVM_AUTO_NETWORK environment variable
5. PENVM_WORLD_FILENAME environment variable
6. <binname>.penvm found in directory containing executable

Source code in penvm/src/client/penvm/client/world.py

class World(BaseObject):
    """World of networks and hosts.

    Configuration is taken from:

    1. `config` keyword argument
    1. `networkstr` keyword argument
    1. `filename` keyword argument
    1. `PENVM_AUTO_NETWORK` environment variable
    1. `PENVM_WORLD_FILENAME` environment variable
    1. `<binname>.penvm` found in directory containing executable
    """

    def __init__(self, **kwargs):
        """Initialize.

        Keyword Args:
            config (dict): Dictionary configuration.
            filename (str): Configuration filename.
            networkstr (str): Network string of machine connection
                strings.
        """
        try:
            super().__init__(kwargs.get("filename"), logger)
            tlogger = self.logger.enter()

            self.config = WorldConfig()
            self.networks = {}
            self.filename = None

            # print(f"{kwargs=}")
            if kwargs.get("config") != None:
                # print("World config")
                self.config.load_config(kwargs["config"])
            elif kwargs.get("networkstr") != None:
                # print("World networkstr")
                self.load_auto_network("default", kwargs["networkstr"])
            elif kwargs.get("filename") != None:
                # print("World filename")
                self.filename = kwargs["filename"]
                self.config.load(self.filename)
            elif os.environ.get("PENVM_AUTO_NETWORK") != None:
                # print("World PENVM_AUTO_NETWORK")
                self.load_auto_network("default", os.environ.get("PENVM_AUTO_NETWORK"))
            elif os.environ.get("PENVM_WORLD_FILENAME") != None:
                # print("World PENVM_WORLD_FILENAME")
                self.filename = os.environ.get("PENVM_WORLD_FILENAME")
                self.config.load(self.filename)
            else:
                # print("World .penvm")
                filename, ext = os.path.splitext(os.path.basename(sys.argv[0]))
                filename = f"{os.path.dirname(sys.argv[0])}/{filename}.penvm"
                # TODO: abspath(filename)?
                if filename and os.path.exists(filename):
                    self.config.load(filename)
                    self.filename = filename

            if self.filename:
                # TODO: is this the right place for this or way to do this?
                self.oid = self.filename
        except Exception as e:
            self.logger.warning(f"EXCEPTION ({e})")
        finally:
            tlogger.exit()

    def get_group_names(self):
        return self.config.get_groups()

    def get_meta(self, name: str) -> Any:
        """Get meta(data) value.

        Returns:
            Value.
        """
        return self.config.get_meta(name)

    def get_network(self, network_name: str = "default") -> "Network":
        """Get network object by name.

        Args:
            network_name: Network name.

        Returns:
            Network.
        """
        network = self.networks.get(network_name)
        if not network:
            netconfig = self.config.get_network(network_name)
            if not netconfig:
                return None
            targets = []
            for target_name in netconfig.get_targets():
                target = self.get_target(target_name)
                if target:
                    targets.append(target)
            network = Network(network_name, targets, netconfig)
            self.networks[network_name] = network
        return network

    def get_networks(self) -> List["Network"]:
        """Get networks.

        Return:
            List of networks.
        """
        return list(self.networks.values())

    def get_network_names(self) -> List[str]:
        """Get network names.

        Return:
            Network names.
        """
        return self.config.get_networks()

    def get_target(self, target_name: str) -> "Target":
        """Get target by name.

        Args:
            target_name: Target name.

        Returns:
            Target.
        """
        return Target(target_name, self.config.get_target(target_name))

    def get_target_names(self) -> List[str]:
        """Get target names.

        Returns:
            List of target names.
        """
        return self.config.get_targets()

    def load_auto_network(self, name: str, networkstr: str):
        """Load auto-scheme (predefined, running) network

        The machine connection strings are used to generate a
        configuration for targets (by machid) and a network.

        Args:
            name: Network name.
            networkstr: Network string.
        """
        try:
            targets = []
            for machconnstr in networkstr.split():
                mcs = MachineConnectionSpec(machconnstr=machconnstr)
                target_config = {
                    "machine-id": mcs.machid,
                    "scheme": "auto",
                    "host": mcs.host,
                    "port": mcs.port,
                    "ssl-profile": mcs.sslprofile,
                }
                targets.append(mcs.machid)
                self.config.add_target(mcs.machid, target_config)
            net_config = {
                "targets": " ".join(targets),
            }
            self.config.add_network(name, net_config)
        except Exception as e:
            traceback.print_exc()

    def shutdown(self):
        """Shut down network.

        Results in machine instances being shut down.
        """
        for network in self.get_networks():
            try:
                network.shutdown()
            except Exception as e:
                pass

__init__(**kwargs)

Initialize.

Other Parameters:

Name	Type	Description
config	dict	Dictionary configuration.
filename	str	Configuration filename.
networkstr	str	Network string of machine connection strings.

Source code in penvm/src/client/penvm/client/world.py

def __init__(self, **kwargs):
    """Initialize.

    Keyword Args:
        config (dict): Dictionary configuration.
        filename (str): Configuration filename.
        networkstr (str): Network string of machine connection
            strings.
    """
    try:
        super().__init__(kwargs.get("filename"), logger)
        tlogger = self.logger.enter()

        self.config = WorldConfig()
        self.networks = {}
        self.filename = None

        # print(f"{kwargs=}")
        if kwargs.get("config") != None:
            # print("World config")
            self.config.load_config(kwargs["config"])
        elif kwargs.get("networkstr") != None:
            # print("World networkstr")
            self.load_auto_network("default", kwargs["networkstr"])
        elif kwargs.get("filename") != None:
            # print("World filename")
            self.filename = kwargs["filename"]
            self.config.load(self.filename)
        elif os.environ.get("PENVM_AUTO_NETWORK") != None:
            # print("World PENVM_AUTO_NETWORK")
            self.load_auto_network("default", os.environ.get("PENVM_AUTO_NETWORK"))
        elif os.environ.get("PENVM_WORLD_FILENAME") != None:
            # print("World PENVM_WORLD_FILENAME")
            self.filename = os.environ.get("PENVM_WORLD_FILENAME")
            self.config.load(self.filename)
        else:
            # print("World .penvm")
            filename, ext = os.path.splitext(os.path.basename(sys.argv[0]))
            filename = f"{os.path.dirname(sys.argv[0])}/{filename}.penvm"
            # TODO: abspath(filename)?
            if filename and os.path.exists(filename):
                self.config.load(filename)
                self.filename = filename

        if self.filename:
            # TODO: is this the right place for this or way to do this?
            self.oid = self.filename
    except Exception as e:
        self.logger.warning(f"EXCEPTION ({e})")
    finally:
        tlogger.exit()

get_meta(name)

Get meta(data) value.

Returns:

Type	Description
Any	Value.

Source code in penvm/src/client/penvm/client/world.py

def get_meta(self, name: str) -> Any:
    """Get meta(data) value.

    Returns:
        Value.
    """
    return self.config.get_meta(name)

get_network(network_name='default')

Get network object by name.

Parameters:

Name	Type	Description	Default
network_name	str	Network name.	'default'

Returns:

Type	Description
Network	Network.

Source code in penvm/src/client/penvm/client/world.py

def get_network(self, network_name: str = "default") -> "Network":
    """Get network object by name.

    Args:
        network_name: Network name.

    Returns:
        Network.
    """
    network = self.networks.get(network_name)
    if not network:
        netconfig = self.config.get_network(network_name)
        if not netconfig:
            return None
        targets = []
        for target_name in netconfig.get_targets():
            target = self.get_target(target_name)
            if target:
                targets.append(target)
        network = Network(network_name, targets, netconfig)
        self.networks[network_name] = network
    return network

get_network_names()

Get network names.

Return

Network names.

Source code in penvm/src/client/penvm/client/world.py

def get_network_names(self) -> List[str]:
    """Get network names.

    Return:
        Network names.
    """
    return self.config.get_networks()

get_networks()

Get networks.

Return

List of networks.

Source code in penvm/src/client/penvm/client/world.py

def get_networks(self) -> List["Network"]:
    """Get networks.

    Return:
        List of networks.
    """
    return list(self.networks.values())

get_target(target_name)

Get target by name.

Parameters:

Name	Type	Description	Default
target_name	str	Target name.	required

Returns:

Type	Description
Target	Target.

Source code in penvm/src/client/penvm/client/world.py

def get_target(self, target_name: str) -> "Target":
    """Get target by name.

    Args:
        target_name: Target name.

    Returns:
        Target.
    """
    return Target(target_name, self.config.get_target(target_name))

get_target_names()

Get target names.

Returns:

Type	Description
List[str]	List of target names.

Source code in penvm/src/client/penvm/client/world.py

def get_target_names(self) -> List[str]:
    """Get target names.

    Returns:
        List of target names.
    """
    return self.config.get_targets()

load_auto_network(name, networkstr)

Load auto-scheme (predefined, running) network

The machine connection strings are used to generate a configuration for targets (by machid) and a network.

Parameters:

Name	Type	Description	Default
name	str	Network name.	required
networkstr	str	Network string.	required

Source code in penvm/src/client/penvm/client/world.py

def load_auto_network(self, name: str, networkstr: str):
    """Load auto-scheme (predefined, running) network

    The machine connection strings are used to generate a
    configuration for targets (by machid) and a network.

    Args:
        name: Network name.
        networkstr: Network string.
    """
    try:
        targets = []
        for machconnstr in networkstr.split():
            mcs = MachineConnectionSpec(machconnstr=machconnstr)
            target_config = {
                "machine-id": mcs.machid,
                "scheme": "auto",
                "host": mcs.host,
                "port": mcs.port,
                "ssl-profile": mcs.sslprofile,
            }
            targets.append(mcs.machid)
            self.config.add_target(mcs.machid, target_config)
        net_config = {
            "targets": " ".join(targets),
        }
        self.config.add_network(name, net_config)
    except Exception as e:
        traceback.print_exc()

shutdown()

Shut down network.

Results in machine instances being shut down.

Source code in penvm/src/client/penvm/client/world.py

def shutdown(self):
    """Shut down network.

    Results in machine instances being shut down.
    """
    for network in self.get_networks():
        try:
            network.shutdown()
        except Exception as e:
            pass