/.venv-3.12/

/cached-cdi.xml

- (recommended) ms-python (VSCode will recommend it when opening and/or running a Python workspace or file typically):

- You must install ms-python extension version 2025.0.0 or earlier of you are actually using Python 3.8, otherwise you will need a later version of Python.

import copy

import sys

import xml.sax

import xml.sax.handler

import xml.sax.xmlreader # for static type hints, autocomplete in this case

from logging import getLogger

from openlcb import precise_sleep

from openlcb.xmldataprocessor import attrs_to_dict

from openlcb.tcplink.tcpsocket import TcpSocket

logger = getLogger(__file__)

else:

logger = getLogger(__name__)

from openlcb.canbus.canphysicallayergridconnect import ( # noqa:E402

from openlcb.canbus.canlink import CanLink # noqa:E402

from openlcb.nodeid import NodeID # noqa:E402

from openlcb.datagramservice import ( # noqa:E402

from openlcb.memoryservice import ( # noqa:E402

sock = TcpSocket()

sock.connect(settings['host'], settings['port'])

physicalLayer = CanPhysicalLayerGridConnect()

physicalLayer.registerFrameReceivedListener(printFrame)

canLink = CanLink(physicalLayer, NodeID(settings['localNodeID']))

complete_data = False

read_failed = False

global complete_data

logger.debug(

f"[{memo.address}] successful read"

f" {MemoryService.arrayToString(memo.data, len(memo.data))}"

"; next = address + 64")

complete_data = True

global read_failed

read_failed = True

"""XML SAX callbacks in a handler object

def __init__(self):

self._chunks = []

self.stack = []

self.cursorCol = 0

self._tmp_space = None # type: int|None

self._tmp_address = None # type: int|None

def startElement(self, name: str, attrs: xml.sax.xmlreader.AttributesImpl):

"""See xml.sax.handler.ContentHandler documentation."""

self.stack.append(name)

if self.cursorCol != 0:

self.print()

self.write(name)

if attrs is not None and attrs:

self.print(" {}".format(attrs_to_dict(attrs)))

def endElement(self, name: str):

"""See xml.sax.handler.ContentHandler documentation."""

content = self._flushCharBuffer().strip()

if self.cursorCol != 0:

self.print()

if content:

self.print('/{} "{}"'.format(name, content))

else:

self.print('/{}'.format(name))

self.stack.pop()

# self.print("/", name)

def write(self, *args, **kwargs):

args = list(args)

if self.cursorCol == 0:

tab = len(self.stack)*" "

self.cursorCol += len(tab)

args.insert(0, tab) # prepend indent

for arg in args:

sys.stdout.write(arg)

self.cursorCol += len(arg)

sys.stdout.flush()

def print(self, *args, **kwargs):

if self.cursorCol == 0: # No indent yet, so use write.

self.write(*args, **kwargs)

print()

else:

print(*args, **kwargs)

self.cursorCol = 0

s = ''.join(self._chunks)

self._chunks.clear()

def characters(self, content: str):

"""Received characters handler.

if not isinstance(content, str):

raise TypeError("Expected str, got {}"

.format(type(content).__name__))

self._chunks.append(content)

def processXML(content: str) :

with open("cached-cdi.xml", 'w') as stream:

# NOTE: Actual caching should key by all SNIP info that could

# affect CDI/FDI: manufacturer, model, and version. Without

# all 3 being present in SNIP, the cache may be incorrect.

stream.write(content)

print(" QUEUE frames : link up...")

physicalLayer.physicalLayerUp()

print(" QUEUED frames : link up...waiting...")

while canLink.pollState() != CanLink.State.Permitted:

# provides incoming data to physicalLayer & sends queued:

physicalLayer.receiveAll(sock, verbose=True)

physicalLayer.sendAll(sock)

if canLink.getState() == CanLink.State.WaitForAliases:

# physicalLayer.receiveAll(sock, verbose=True)

physicalLayer.sendAll(sock)

# ^ prevent assertion error below, proceed to send.

if canLink.pollState() == CanLink.State.Permitted:

break

assert canLink.getWaitForAliasResponseStart() is not None, \

("openlcb didn't send the 7,6,5,4 CID frames (state={})"

.format(canLink.getState()))

precise_sleep(.02)

print(" SENT frames : link up")

time.sleep(.21)

# ^ 200ms: See section 6.2.1 of CAN Frame Transfer Standard

# (CanLink.State.Permitted will only occur after that, but waiting

# now will reduce output & delays below in this example).

while canLink.getState() != CanLink.State.Permitted:

print("Waiting for connection sequence to complete...")

# This delay could be .2 (per alias collision), but longer to

# reduce console messages:

time.sleep(.5)

print("Requesting memory read. Please wait...")

print()

print("This example will exit on failure or complete data.")

while not complete_data and not read_failed:

# In this example, requests are initiate by the

# memoryRead thread, and receiveAll actually

# receives the data from the requested memory space (CDI in this

# case) and offset (incremental position in the file/data,

# incremented by this example's memoryReadSuccess handler).

count = 0

count += physicalLayer.receiveAll(sock)

count += physicalLayer.sendAll(sock)

if canLink.nodeIdToAlias != previous_nodes:

print("nodeIdToAlias updated: {}".format(canLink.nodeIdToAlias))

if count < 1:

precise_sleep(.01)

# else skip sleep to avoid latency (port already delayed)

if canLink.nodeIdToAlias != previous_nodes:

previous_nodes = copy.deepcopy(canLink.nodeIdToAlias)

physicalLayer.physicalLayerDown()

if read_failed:

print("Read complete (FAILED)")

else:

print("Read complete (OK)")