flync.core

flync.core.annotations

external

External field annotations and output strategies.

class NamingStrategy(*values)

Bases: IntEnum

The strategy on how the external field file/folder will be named.

class OutputStrategy(*values)

Bases: IntFlag

The strategy on how an external field will be generated.

class External(path: str | None = None, root: str | None = None, output_structure: OutputStrategy = <OutputStrategy.AUTO: 1>, naming_strategy: NamingStrategy = NamingStrategy.AUTO)

Bases: object

Indicates this field is loaded from a separate location.

implied

Provides annotations for fields that are automatically derived rather than manually set.

class ImpliedStrategy(*values)

Bases: IntEnum

The strategy on how an implied field will be calculated.

class Implied(strategy: ImpliedStrategy = ImpliedStrategy.AUTO)

Bases: object

Indicates this field is implied instead of loaded/generated.

flync.core.base_models

base_models

Base Model that is used by FLYNC Model classes.

class FLYNCBaseModel

Bases: BaseModel

Base Model that is used by FLYNC Model classes.

model_config = {'extra': 'forbid', 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(_FLYNCBaseModel__context)

Override this method to perform additional initialization after __init__ and model_construct. This is useful if you want to do some validation that requires the entire model to be initialized.

model_dump(**kwargs)

Override pydantics model_dump to dump with defaults.

flync.core.datatypes

class Datatype

Bases: FLYNCBaseModel

Base class of every datatype.

Parameters

namestr

Unique name of the datatype.

descriptionstr, optional

Human-readable description of the datatype.

typestr

Discriminator identifying the concrete datatype kind.

endiannessLiteral[“BE”, “LE”], optional

Byte order used for encoding multibyte values. Defaults to big-endian (“BE”).

member_namestr, optional

When this datatype is stored as a struct member, this field holds the member’s name within the struct (which may differ from the type’s own name). None when the datatype is not a struct member or when member name equals the type name.

---

class IPv4AddressEntry

Bases: FLYNCBaseModel

Represents an IPv4 address entry for a network interface.

Parameters

addressIPv4Address

The IPv4 address. “0.0.0.0” means dynamic.

ipv4netmaskIPv4Address

The subnet mask in IPv4 format.

class IPv6AddressEntry

Bases: FLYNCBaseModel

Represents an IPv6 address entry for a network interface.

Parameters

addressIPv6Address

The IPv6 address. “::” means dynamic.

ipv6prefixint

The prefix length (0-128).

---

class MACAddressEntry

Bases: FLYNCBaseModel

Represents an MAC address entry for a network interface.

Parameters:

• address (str):

  Source MAC address to filter by. Format: “xx:xx:xx:xx:xx:xx”

• macmask (str):

  The mask in MAC format. Format: “xx:xx:xx:xx:xx:xx”

class MACAddressUnicast

Bases: MACAddressEntry

Represents a Unicast MAC address entry for a network interface.

class MACAddressMulticast

Bases: MACAddressEntry

Represents a Multicast MAC address entry for a network interface.

---

class BitRange

Bases: Datatype

Represents a range of bits in a signal or data array.

Parameters

namestr

Unique name of the datatype.

descriptionstr, optional

Human-readable description of the datatype.

typestr

Datatype discriminator inherited from Datatype.

endiannessLiteral[“BE”, “LE”], optional

Byte order used for encoding multibyte values. Defaults to big-endian (“BE”).

startint

Starting bit position (inclusive).

endint

Ending bit position (inclusive).

---

class ValueRange

Bases: FLYNCBaseModel

Defines an inclusive range of integer values.

This datatype is typically used to express valid numeric intervals for parameters, identifiers, or signal values.

Parameters

from_valueint

Lower bound of the range (inclusive).

to_valueint

Upper bound of the range (inclusive).

class ValueTable

Bases: FLYNCBaseModel

Represents a table of values with an associated description.

Parameters

num_valueint

The numeric value associated with this table entry.

descriptionstr

Human-readable description of the table entry.

---

class Ethertype

Bases: Enum

EtherType values relevant for automotive and TSN Ethernet networks.

Each member maps a protocol to its IEEE-registered EtherType (the 16-bit field in the Ethernet header that identifies the upper-layer protocol).

Supported EtherTypes:

Name	Hex	Description
ARP	0x0806	Address Resolution Protocol
ASAM_CMP	0x99FE	ASAM Capture Module Protocol
AVTP	0x22F0	IEEE 1722 Audio Video Transport Protocol
ETH_FLOWCTRL	0x8808	Ethernet flow control (PAUSE / IEEE 802.3x)
HSR	0x892F	High-availability Seamless Redundancy (IEC 62439-3)
IPv4	0x0800	Internet Protocol version 4
IPv6	0x86DD	Internet Protocol version 6
LLDP	0x88CC	Link Layer Discovery Protocol (IEEE 802.1AB)
MACsec	0x88E5	MAC Security (IEEE 802.1AE)
PRP	0x88FB	Parallel Redundancy Protocol supervision (IEC 62439-3)
PTP	0x88F7	Precision Time Protocol / gPTP (IEEE 1588 / 802.1AS)
QinQ	0x88A8	Outer VLAN tag (IEEE 802.1ad)
SRP	0x22EA	Stream Reservation Protocol (IEEE 802.1Qat)
VLAN	0x8100	Inner VLAN tag (IEEE 802.1Q)
WAKE_ON_LAN	0x0842	Wake-on-LAN magic packet (ECU wakeup)

Usage:

The Ethertype can be specified in three ways:

• By enum member: Ethertype.AVTP

• By name string: "AVTP"

• By hex value: 0x22F0 or "0x22F0"

Example:

from flync.core.datatypes import Ethertype

# All of these are equivalent:
et1 = Ethertype.AVTP
et2 = Ethertype(0x22F0)
et3 = Ethertype["AVTP"]

# When used in a model, serialization returns hex format:
# {"ethertype": "0x22F0"}

Use in Pydantic models:

eth_type: Annotated[Ethertype, PlainSerializer(serialize_ethertype), BeforeValidator(validate_ethertype_input)]

---

flync.core.utils

base_utils

Base Utils that can be useful throughout the whole FLYNC Library and toolchain.

read_yaml(path: str | PathLike | Path)

Read a YAML file.

Args:

path (str | os.PathLike | Path): Path to the YAML file

Raises:

err_fatal: If path is not for YAML file.

Returns:

Any: Retrieved data from YAML.

get_yaml_paths(base_path: str | PathLike) → list

Collect absolute paths to yaml files from a base_path.

Args:

base_path (str | os.PathLike): Base Path to FLYNC Config.

Returns:

list: List of absolute file paths to yaml files.

is_mac_address(input: str) → Tuple[bool, str]

Helper to check if an input is a valid MAC address based on pydantic validator.

Args:

input (str): input string that should be checked.

Returns:

bool: Returns the result of check as a boolean as well as a message that could be used in logging or exception handling. If the result boolean is true, the provided input is a MAC address. If the boolean is false, it is not.

is_mac_unicast(input: str) → Tuple[bool, str]

Helper to check if a MAC address is unicast. Unicast if first byte’s least significant bit is 0.

Args:

input (str): input string that should be checked.

Returns:

Union[bool, str]: Returns the result of check as a boolean as well as a message that could be used in logging or exception handling. If the result boolean is true, the provided input is a unicast MAC address. If the boolean is false, it is not.

is_mac_multicast(input: str) → Tuple[bool, str]

Method to check if a MAC address is multicast. Multicast if first byte’s least significant bit is 1.

Args:

input (str): input string that should be checked.

Returns:

Union[bool, str]: Returns the result of check as a boolean as well as a message that could be used in logging or exception handling. If the result boolean is true, the provided input is a MAC multicast address. If the boolean is false, it is not.

is_ip_address(input: IPv4Address | IPv6Address | str) → Tuple[bool, str]

Helper to check if an input is a valid IP address.

Args:

input (IPv4Address | IPv6Address | str): input string that should be checked.

Returns:

bool: Returns the result of check as a boolean as well as a message that could be used in logging or exception handling. If the result boolean is true, the provided input is an IP address. If the boolean is false, it is not.

is_ip_multicast(input: IPv4Address | IPv6Address | str) → Tuple[bool, str]

Method to check if a string is an IP multicast address.

Args:

input (IPv4Address | IPv6Address): input string that should be checked.

Returns:

Union[bool, str]: Returns the result of check as a boolean as well as a message that could be used in logging or exception handling. If the result boolean is true, the provided input is an IP multicast address. If the boolean is false, it is not.

get_duplicates_in_list(input: list) → list

Find duplicates in a list.

Args:

input (list): a list where duplicates are suspected.

Returns:

list: returns a list of the duplicates that were found.

check_obj_in_list(obj, list)

Helper function: To check if the object is in the list or not

deep_iter(obj, stop_class)

Helper function: To deeply iterate through an object and its children. The iteration will stop if an object of type stop_class is reached.

Args:

obj: The object to iterate through. stop_class: The class type at which the iteration should stop.

Returns:

Generator: A generator that yields objects of the specified type.

find_all(base, target_class: Type[T]) → list[T]

Helper to get all fields of certain type in nested data.

Args:

base: The base object to start searching from. target_class: The class to search for.

Returns:

list: A list of all instances of the target class found within the base object.

common_validators

Common Validators are validation methods that are used throughout the whole FLYNC model. The Validators either raise minor, major or fatal errors as pydantic usage proposes.

validate_vlan_id(value)

Validate a VLAN identifier.

None is treated as untagged and returned unchanged. Values in the range 0-4094 are accepted as-is. The reserved value 4095 is accepted but emits a warning via warn(). Anything outside 0-4095 raises a minor validation error.

validate_or_remove(label: str, field_type: Any, severity: str = 'minor')

Factory that returns a BeforeValidator for sub-model fields.

Use inside Annotated to pre-validate a field before Pydantic processes it. If the raw data fails validation all sub-errors are packed into a single error.

• "minor" severity: the field is removed and the parent model still loads without it.

  The message says “Removing {label}…”.

• "major" severity: the parent model will fail regardless (the field is required).

  The message reports the validation failure without implying graceful removal.

The parent object’s name field is included in the error message when available via info.data.

Parameters

labelstr

Human-readable field label used in the error message.

field_typeAny

Pydantic-compatible type to validate the data against.

severitystr, optional

Error severity — "minor" (default) or "major".

Returns

Callable

A two-argument validator (data, info) ready for use with BeforeValidator.

validate_list_items_and_remove(label: str, item_type: Any, severity: str = 'minor')

Validate each item in a list individually, removing only invalid entries.

Unlike validate_or_remove(), which discards the entire list when any item fails, this validator keeps valid items and removes only those that fail validation. Per-item errors are forwarded to the _validation_warnings channel so they appear in the final error report even though the model continues building with the remaining valid items.

Use inside Annotated as a BeforeValidator.

Parameters

labelstr

Human-readable field label used in error messages.

item_typeAny

Pydantic-compatible type for each individual list item.

severitystr, optional

Error severity for removed items — "minor" (default) or "major".

Returns

Callable

A two-argument validator (data, info) ready for use with BeforeValidator.

validate_mac_unicast(input: str) → str

Custom Validator for Unicast MAC addresses.

Args:

input (str): MAC address to validate.

Raises:

err_minor: Input is not a Unicast address based on the expected format.

Returns:

Any: Input is handed over.

validate_mac_multicast(input: str) → Any

Custom Validator for Multicast MAC addresses.

Args:

input (str): MAC address to validate.

Raises:

err_minor: Input is not a Multicast address based on the expected format.

Returns:

Any: Input is handed over.

validate_ip_multicast(input: IPv4Address | IPv6Address | str) → Any

Custom Validator for Multicast IP addresses.

Args:

input (IPv4Address | IPv6Address): IP address to validate.

Raises:

err_minor: Input is not a Multicast address based on the expected format.

Returns:

Any: Input is handed over.

validate_any_multicast_address(input: IPv4Address | IPv6Address | str) → Any

Custom Validator for Multicast MAC or IP addresses.

Args:

input (IPv4Address | IPv6Address | str): IP address or MAC Address to validate.

Raises:

err_minor: The address is not a multicast address.

Returns:

Any: Input is handed over.

validate_multicast_list_only_ip(input_list: list)

Custom Validator for a list of Multicast IP addresses.

Args:

input_list (list): List of only Multicast IPs.

Raises:

err_minor: Any of the addresses in the list is not an IP multicast address.

validate_multicast_list(input_list: list)

Custom Validator for a list of Multicast MAC or IP addresses.

Args:

input_list (list): List of Multicast IPs and MACs.

Raises:

err_minor: Any of the addresses in the list is not a multicast address.

validate_ingress_streams_fields(streams, location: str)

Raise err_minor if any stream carries an ipv or ats value.

location is a human-readable label such as "compute node" or "controller interface" used in the error message.

validate_vlan_ids_unique(virtual_interfaces, name: str)

Raise err_major if any VLAN ID appears more than once.

validate_list_items_unique(input_list: list, list_label: str | None = None) → list

Custom Validator for a list of items where every item should be unique.

Args:

input_list (list): List of items.

list_label(str): Add an optional label to the error message.

Raises:

err_major: List contains duplicates.

Returns:

list: Input is handed over.

validate_cbs_idleslopes_fit_portspeed(traffic_classes: list, port_speed: int)

Custom Validator for a list of Traffic Classes to check conformity to MII/MDI speed.

Args:

traffic_classes (list): List of element type TrafficClass.

port_speed (int): MII or MDI speed of the port.

Raises:

err_major: The sum of idleslopes of all shapers on one port must be equal or lower than the port speed.

Returns:

list: Return list of traffic classes as received.

validate_optional_mii_config_compatibility(comp1, comp2, id)

Custom validator for optional MII configuration compatibility between two components.

Args:

comp1 (object): First component that may contain a mii_config attribute.

comp2 (object): Second component that may contain a mii_config attribute.

id (Any): Identifier of the connection (used only in error messages).

Raises:

err_major: One component has an MII config while the other does not.

err_major: Both components have an MII config but the mode values are identical. The modes must differ.

err_major: Both components have an MII config but the speed values are different.

err_major: Both components have an MII config but the type values are different.

validate_compulsory_mii_config_compatibility(comp1, comp2, id)

Validator that enforces a mandatory MII configuration on both components and then checks optional compatibility.

Args:

comp1 (object): First component. Must have mii_config.

comp2 (object): Second component. Must have mii_config.

id (Any): Identifier of the connection (used only in error messages).

Raises:

err_major: Either component is missing a required MII configuration.

err_major: Propagated from validate_optional_mii_config_compatibility() when the optional checks fail.

validate_htb(comp, speed)

Validator that checks an HTB (Hierarchical Token Bucket) configuration against the physical link speed.

Args:

comp (object): Component that owns an htb attribute with child_classes.

speed (int): Link speed of the interface (same unit as the HTB rates).

Raises:

err_major: The sum of the rate values of all child classes exceeds the provided speed.

validate_macsec(comp1, comp2, id)

Validator for MACsec configuration compatibility between two components.

Args:

comp1 (object): First component: May contain a macsec_config.

comp2 (object): Second component: May contain a macsec_config.

id (Any): Identifier of the connection (used only in error messages).

Raises:

err_major: One component has a MACsec config while the other does not.

err_major: MKA (Key Agreement) enabled state differs between the two components.

err_major: macsec_mode differs between the two components.

validate_gptp(comp1, comp2, id)

Validator that checks gPTP (generic Precision Time Protocol) configuration compatibility between two components.

Args:

comp1 (object): First component. May contain a ptp_config.

comp2 (object): Second component. May contain a ptp_config.

id (Any): Identifier of the connection (used only in error messages).

Raises:

err_major: PTP configuration present on one side only.

err_major: Mismatch of the cmlds_linkport_enabled flag between the two components.

err_major: Propagated from validate_gptp_domains() when domain level checks fail.

validate_gptp_domains(comp1, comp2, ptp1, ptp2, id)

Helper that validates matching PTP domains and sync-config types between two components.

Args:

comp1 (object): First component (source of ptp1).

comp2 (object): Second component (source of ptp2).

ptp1 (object): ptp_config of comp1.

ptp2 (object): ptp_config of comp2.

id (Any): Identifier of the connection (used only in error messages).

Raises:

err_major: A domain present in ptp1 is missing in ptp2.

err_major: The sync_config.type of a matching domain is identical on both sides (they must differ for a valid configuration).

validate_elements_in(subset: Iterable[Any], superset: Iterable[Any], msg: str | None = None)

Custom Validator that checks if every element in subset appears at least once in superset. E.g. Validate if port_name is in switch_port_names.

Args:

subset (Iterable[Any]): Subset where elements are expected to be in superset.

superset (Iterable[Any]): Reference set.

Returns:

Iterable[Any]: Return subset as received.

check_prio_unique(traffic_classes)

Check if the traffic class prios are unique across various traffic classes in a controller interface or switch.

check_pcps_different(traffic_classes)

Check if the PCPs are different across traffic classes.

check_ipvs_unique(traffic_classes)

Check if ipvs across traffic classes are unique.

validate_traffic_classes(traffic_classes)

Validate the traffic classes in a controller interface and switch to find out if a pcp, ipv or traffic class prio is reused or not.

none_to_empty_list(v, info=None)

Make the field defined as optional [] if accidentally declared by the user as None.

collect_bit_ranges(items: Iterable[Any], get_range: Callable[[Any], Tuple[str, int, int] | None]) → List[Tuple[str, int, int]]

Build a list of (name, start_bit, end_bit_exclusive) ranges from items.

get_range(item) is called for every entry and should return either a (name, start_bit, end_bit_exclusive) tuple or None when the item is unplaced and should be skipped (e.g. a SignalInstance with no bit_position).

check_bit_ranges_within(context: str, ranges: Iterable[Tuple[str, int, int]], max_bits: int) → None

Raise err_minor() when any range extends past max_bits.

context is a human-readable label of the container (PDU or frame name) used in the error message.

check_bit_ranges_no_overlap(context: str, ranges: List[Tuple[str, int, int]]) → None

Raise err_minor() when any two ranges in ranges intersect.

Ranges are half-open [start, end); two ranges overlap when start_a < end_b and start_b < end_a. context is included in the error message to identify the enclosing PDU or frame.

validate_value_input_format(data: dict) → dict

Validating combinations of ‘value’, ‘from_value’ and ‘to_value’.

exceptions

Defines custom pydantic errors and the FLYNC error-id scheme.

class Severity(*values)

Bases: StrEnum

String enum that represents errors severity levels

class Category(*values)

Bases: IntEnum

Integer enum to represent categories for errors

module_code_for(dotted_module: str) → str

Resolve a module code from the KEY declared in the nearest enclosing package.

Walks the package chain (flync.model.flync_4_ecu.port -> flync_4_ecu -> ECU). Falls back to GEN for the top-level model and version migrators, CMN otherwise.

compose_error_id(severity: Severity, module_code: str, category: Category | None, error_number: str | None) → str

Assemble FLYNC-<MODULE>-<SEVERITY>-<CATEGORY>-<NUMBER>; unset category/number render as UNC/000.

warn(msg: str, *, category: Category | None = None, error_number: str | None = None, **ctx) → None

Record a validation warning without raising a validation error.

The message is appended to the active warning list (set up by validate_with_policy) and will be returned alongside load_errors so that it appears in the warnings table. If called outside a validate_with_policy context the call is silently ignored.

Parameters

msgstr

Human-readable warning message.

categoryCategory, optional

Semantic category used to compose the error id.

error_numberstr, optional

Globally unique zero-padded number identifying this call site.

ctxdict

Context arguments that define key-value pairs to fill the placeholders in msg.

err_minor(msg: str, *, category: Category | None = None, error_number: str | None = None, **ctx) → PydanticCustomError

Factory that returns PydanticCustomError with type minor.

Parameters

msgstr

Error message that may contain placeholders

categoryCategory, optional

Semantic category used to compose the error id.

error_numberstr, optional

Globally unique zero-padded number identifying this call site.

ctxdict

Context arguments that define key-value pairs to fill the placeholders in msg

Returns

PydanticCustomError

err_major(msg: str, *, category: Category | None = None, error_number: str | None = None, **ctx) → PydanticCustomError

Factory that returns PydanticCustomError with type major.

Parameters

msgstr

Error message that may contain placeholders

categoryCategory, optional

Semantic category used to compose the error id.

error_numberstr, optional

Globally unique zero-padded number identifying this call site.

ctxdict

Context arguments that define key-value pairs to fill the placeholders in msg

Returns

PydanticCustomError

err_fatal(msg: str, *, category: Category | None = None, error_number: str | None = None, **ctx) → PydanticCustomError

Factory that returns PydanticCustomError with type fatal.

Parameters

msgstr

Error message that may contain placeholders

categoryCategory, optional

Semantic category used to compose the error id.

error_numberstr, optional

Globally unique zero-padded number identifying this call site.

Returns

PydanticCustomError

exceptions_handling

Provides utilities for validating models and handling errors.

is_semantic_validation_error(err: ErrorDetails) → bool

Return True if err is a user-raised semantic validation error.

User code raises semantic errors via err_major(), err_minor(), or err_fatal(); their type is exactly "major" / "minor" / "fatal" and ctx carries the formatter kwargs only. A native Pydantic mismatch rewrapped by _wrap_native_error() shares the "major" type but always sets ctx["sub_errors"] — that key distinguishes structural mismatches from semantic failures.

resolve_alias(model: type[BaseModel], field_name: str) → str

Return the YAML key used for a Pydantic field, considering alias.

get_name_by_alias(model: type[BaseModel], alias: str)

Return the Python field name that corresponds to the given alias.

Parameters

modeltype[BaseModel]

The Pydantic model class to search.

aliasstr

The alias to look up.

Returns

str

The Python attribute name whose alias matches alias.

Raises

KeyError

If no field with the given alias is found.

safe_yaml_position(node: Any, loc: tuple, model: type[BaseModel] | None = None) → Tuple[int | None, int | None]

Given a ruamel.yaml node and a Pydantic loc tuple, return (line, column). Falls back gracefully if key/item is missing.

errors_to_init_errors(errors: List[ErrorDetails], model: type[BaseModel] | None = None, yaml_data: object | None = None, yaml_path: str | None = None) → List[InitErrorDetails]

Convert Pydantic validation errors into InitErrorDetails for re-raising.

Optionally enriches each error with YAML source location information when model and yaml_data are provided, and with the file path when yaml_path is provided.

Parameters

errorsList[ErrorDetails]

The list of errors to convert.

modeltype[BaseModel], optional

The Pydantic model class used to resolve field aliases for YAML position look-ups.

yaml_dataobject, optional

The parsed ruamel.yaml AST of the document, used together with model to locate the error position within the file.

yaml_pathstr, optional

The workspace-relative file path to embed in each error’s context as yaml_path.

Returns

List[InitErrorDetails]

The converted errors, ready to be passed to ValidationError.from_exception_data.

delete_at_loc(data: Any, loc: Tuple)

Helper function to remove the key/item from original object by loc(path to an element within the object).

Parameters

dataAny

Data to remove the item from. Will be mutated.

locTuple

Path to the location of item to remove.

get_unique_errors(errors: List[ErrorDetails]) → List[ErrorDetails]

A function to get the list of unique errors.

Parameters

errors: List[ErrorDetails]

The list of pydantic’s error details

Returns

List[ErrorDetails]

validate_with_policy(model: Type[FLYNCBaseModel], data: Any, path) → Tuple[FLYNCBaseModel | None, List[ErrorDetails]]

Helper function to perform model validation from the given data, collect errors with different severity and perform action based on severity.

For minor/major errors the offending field is removed from the working data via delete_at_loc() and validation is retried, so that the model can still be constructed without the invalid field. The loop continues until either validation succeeds, a fatal error is encountered, or no further progress can be made (all error locations already removed).

Parameters

modelType[FLYNCBaseModel]

Flync model class.

dataAny

Data to validate and instantiate the model with.

Returns

Tuple[Optional[FLYNCBaseModel], List]

Tuple with optional model instance and list of errors.

Raises

ValidationError