T_SFLU_OPODRFPMI

wield.control.SFLU.test.T_SFLU_OPODRFPMI

These tests demonstrate a heierachrical way to build connection graphs for optical systems of mirrors and beamsplitters.

pytest-html report

Functions

T_SFLU_OPODRFPMI_build_show()

Show a graph reduction using networkx+tikz

T_SFLU_OPO_build_show()

Show a graph reduction using networkx+tikz

build_OPO()

This is a pytest needing documentation

Details

T_SFLU_OPODRFPMI_build_show()[source][github]

Show a graph reduction using networkx+tikz

code
docstring
"""
Show a graph reduction using networkx+tikz
"""

  1def T_SFLU_OPODRFPMI_build_show():
  2
  3    # opo['Yetm'].node_angle["fr.o"] = -45
  4    # opo['Yitm'].node_angle["bk.o"] = -45
  5    # opo['srm'].node_angle["bk.o"] = -45
  6    # opo['srm'].edges["bk.i", "bk.i.exc"] = "sec_to"
  7    # opo['srm'].edges["bk.o.tp", "bk.o"] = "sec_fr"
  8    # opo['srm'].edge_handedness["bk.o.tp", "bk.o"] = "r"
  9    opo = build_OPO()
 10
 11    ifo = optics.GraphElement()
 12    ifo.subgraph_add(
 13        'prm', optics.LossyBasisMirror(),
 14        translation_xy=(-25, 0),
 15        rotation_deg=180,
 16    )
 17    ifo.subgraph_add(
 18        'Xitm', optics.LossyBasisMirror(),
 19        translation_xy=(25, 0),
 20        rotation_deg=180,
 21    )
 22    ifo.subgraph_add(
 23        'Xetm', optics.LossyMirror(),
 24        translation_xy=(55, 0),
 25        rotation_deg=0
 26    )
 27    ifo.subgraph_add(
 28        'Yitm', optics.LossyBasisMirror(),
 29        translation_xy=(0, 25),
 30        rotation_deg=90+180,
 31    )
 32    ifo.subgraph_add(
 33        'Yetm', optics.LossyMirror(),
 34        translation_xy=(0, 55),
 35        rotation_deg=90,
 36    )
 37    ifo.subgraph_add(
 38        'bs', optics.BeamSplitter(),
 39        translation_xy=(0, 0),
 40        rotation_deg=0
 41    )
 42    ifo.subgraph_add(
 43        'opo', opo,
 44        translation_xy=(0, -60),
 45        rotation_deg=-90
 46    )
 47    ifo.subgraph_add(
 48        'srm', optics.LossyBasisMirror(),
 49        translation_xy=(+30, -40),
 50        rotation_deg=0,
 51    )
 52    ifo.edges.update({
 53        ("prm.fr.i",  "bs.frA.o"): "prc.tau",
 54        ("bs.frA.i",  "prm.fr.o"): "prc.tau",
 55
 56        ("opo.M1.frA.i", "bs.bkB.o"): "sec.tau",
 57        ("bs.bkB.i", "opo.M1.frA.o"): "sec.tau",
 58        ("srm.fr.i", "opo.M1.frB.o"): "sec.tau",
 59        ("opo.M1.frB.i", "srm.fr.o"): "sec.tau",
 60
 61        ("Yitm.bk.i", "bs.frB.o"): 'BSY.tau',
 62        ("bs.frB.i",  "Yitm.bk.o"): 'BSY.tau',
 63
 64        ("Xitm.bk.i", "bs.bkA.o"): 'BSX.tau',
 65        ("bs.bkA.i",  "Xitm.bk.o"): 'BSX.tau',
 66
 67        ("Xetm.fr.i", "Xitm.fr.o"): "XARM.tau",
 68        ("Xitm.fr.i", "Xetm.fr.o"): "XARM.tau",
 69
 70        ("Yetm.fr.i", "Yitm.fr.o"): "YARM.tau",
 71        ("Yitm.fr.i", "Yetm.fr.o"): "YARM.tau",
 72    })
 73    ifo['Yetm'].node_angle["fr.o"] = -45
 74    ifo['Yitm'].node_angle["bk.o"] = -45
 75    ifo['srm'].node_angle["bk.o"] = -45
 76    ifo['srm'].edges["bk.i", "bk.i.exc"] = "sec_to"
 77    ifo['srm'].edges["bk.o.tp", "bk.o"] = "sec_fr"
 78    ifo['srm'].edge_handedness["bk.o.tp", "bk.o"] = "r"
 79
 80    ifo['srm'].locations["bk.i.exc"] = (15, -10)
 81    ifo['srm'].locations["bk.o.tp"] = (15, 10)
 82
 83    ifo['prm'].edges["bk.i", "bk.i.exc"] = "1"
 84    ifo['prm'].edges["bk.o.tp", "bk.o"] = "1"
 85    ifo['prm'].node_angle["bk.i.exc"] = +45
 86
 87    ifo['prm'].locations["bk.i.exc"] = (15, -10)
 88    ifo['prm'].locations["bk.o.tp"] = (15, 10)
 89    ifo['prm'].edges["fr.i.tp", "fr.i"] = "1"
 90    ifo['prm'].locations["fr.i.tp"] = (-5, 12)
 91
 92    ifo['Xetm'].edges["fr.i.tp", "fr.i"] = "1"
 93    ifo['Xetm'].edges["fr.o", "fr.o.exc"] = "1"
 94    ifo['Xetm'].locations["fr.i.tp"] = (-5, 15)
 95    ifo['Xetm'].locations["fr.o.exc"] = (-5, -15)
 96
 97    ifo['Yetm'].edges["fr.i.tp", "fr.i"] = "1"
 98    ifo['Yetm'].edge_handedness["fr.i.tp", "fr.i"] = "r"
 99    ifo['Yetm'].edges["fr.o", "fr.o.exc"] = "1"
100    ifo['Yetm'].locations["fr.i.tp"] = (-5, 15)
101    ifo['Yetm'].locations["fr.o.exc"] = (-5, -15)
102
103    ifo['Xitm'].edges["bk.i.tp", "bk.i"] = "1"
104    ifo['Xitm'].locations["bk.i.tp"] = (5, -12)
105    ifo['Yitm'].edges["bk.i.tp", "bk.i"] = "1"
106    ifo['Yitm'].locations["bk.i.tp"] = (5, -12)
107
108    sflu = SFLU.SFLU(
109        edges=ifo.build_edges(),
110        derivatives=[
111        ],
112        reduce_list=reduce_list,
113        graph=True,
114    )
115    # match=False allows a reduced input/output set
116    # sflu.graph_nodes_pos(ifo.build_locations(), match=True)
117    ifo.update_sflu(sflu)
118    G1 = sflu.G.copy()
119
120    if True:
121        yamlstr = sflu.convert_self2yamlstr()
122        # print(yamlstr)
123        with open(tjoin('DRFPMI.yaml'), 'w') as F:
124            F.write(yamlstr)
125        sflu = SFLU.SFLU.convert_yamlstr2self(yamlstr)
126    #sflu.graph_nodes_pos(DRFPMI_locs, match=True)
127
128    print('inputs: ', sflu.inputs)
129    print('outputs: ', sflu.outputs)
130    print('nodes: ', sflu.nodes)
131    # dprint('edges: ', sflu.edges)
132
133    print('nodes')
134    #print(sflu.graph_nodes_repr())
135    #G1 = sflu.G.copy()
136    sflu.graph_reduce_auto_pos(lX=-10, rX=+10, Y=0, dY=-2)
137    #sflu.reduce(*reduce_list)
138    #assert(not sflu.nodes)
139    #sflu.graph_reduce_auto_pos_io(lX=-30, rX=+30, Y=10, dY=-2)
140    #G2 = sflu.G.copy()
141    dprint(G1.edges)
142
143    nx2tikz.dump_pdf(
144        [
145            G1,
146            #G2,
147        ],
148        fname = tjoin('testG.pdf'),
149        texname = tjoin('testG.tex'),
150        # preamble = preamble,
151        scale='10pt',
152    )
pytest information

This code is wrapped in a pytest function using conventions detailed in Pytest Conventions. The full name of this test, as known by the documentation, is:

wield.control.SFLU.test.T_SFLU_OPODRFPMI.T_SFLU_OPODRFPMI_build_show

The full name is useful when building documentation, to link a reference to this page using :func:`name`, or directly include it with an autofunction directive. The collapse nodes below show every instance of the test run. There may only be one, but if the test was run multiple times through pytest parametrizations, the list can be longer.

T_SFLU_OPODRFPMI_build_show
output
inputs:  {'Yetm.bkL.i', 'srm.frL.i', 'opo.SQZ.frL.i', 'Xitm.frL.i', 'srm.bk.i.exc', 'opo.M2.bkB.i', 'Yetm.bk.i', 'Xetm.bk.i', 'Xetm.bkL.i', 'Xetm.fr.o.exc', 'Xitm.bkL.i', 'prm.bkL.i', 'Yetm.fr.o.exc', 'prm.bk.i.exc', 'Yetm.frL.i', 'opo.SQZ.bkL.i', 'Xetm.frL.i', 'prm.frL.i', 'opo.M2.bkA.i', 'Yitm.frL.i', 'srm.bkL.i', 'Yitm.bkL.i'}
outputs:  {'opo.M2.bkA.o', 'Yetm.fr.i.tp', 'prm.bk.o.tp', 'Xitm.bk.i.tp', 'srm.bk.o.tp', 'Xetm.fr.i.tp', 'Yitm.bk.i.tp', 'Xetm.bk.o', 'Yetm.bk.o', 'opo.M2.bkB.o', 'prm.fr.i.tp'}
nodes:  {'opo.M2.frB.i', 'Xetm.fr.i', 'opo.SQZ.bk.o', 'prm.bk.o', 'prm.fr.i', 'Xitm.bk.o', 'prm.bk.i', 'bs.frA.o', 'bs.bkA.o', 'bs.bkB.i', 'opo.M2.frB.o', 'opo.M1.frB.i', 'srm.bk.o', 'bs.frA.i', 'Yitm.bk.i', 'Xetm.fr.o', 'Xitm.fr.i', 'Yetm.fr.i', 'opo.M1.frB.o', 'opo.M2.frA.o', 'Yitm.fr.i', 'opo.M1.frA.o', 'bs.frB.o', 'opo.SQZ.bk.i', 'Yitm.fr.o', 'opo.M1.bkA.o', 'srm.bk.i', 'opo.M2.frA.i', 'opo.SQZ.fr.i', 'bs.frB.i', 'Xitm.fr.o', 'srm.fr.i', 'opo.M1.bkA.i', 'Xitm.bk.i', 'Yetm.fr.o', 'srm.fr.o', 'prm.fr.o', 'Yitm.bk.o', 'opo.SQZ.fr.o', 'bs.bkA.i', 'opo.M1.bkB.i', 'opo.M1.frA.i', 'opo.M1.bkB.o', 'bs.bkB.o'}
nodes
Captured stderr call
OutEdgeView([('prm.bk.i', 'prm.fr.o'), ('prm.bk.i', 'prm.bk.o'), ('prm.fr.o', 'bs.frA.i'), ('prm.fr.i', 'prm.bk.o'), ('prm.fr.i', 'prm.fr.o'), ('prm.fr.i', 'prm.fr.i.tp'), ('prm.bk.o', 'prm.bk.o.tp'), ('prm.frL.i', 'prm.fr.o'), ('prm.bkL.i', 'prm.bk.o'), ('prm.bk.i.exc', 'prm.bk.i'), ('Xitm.bk.i', 'Xitm.fr.o'), ('Xitm.bk.i', 'Xitm.bk.o'), ('Xitm.bk.i', 'Xitm.bk.i.tp'), ('Xitm.fr.o', 'Xetm.fr.i'), ('Xitm.fr.i', 'Xitm.bk.o'), ('Xitm.fr.i', 'Xitm.fr.o'), ('Xitm.bk.o', 'bs.bkA.i'), ('Xitm.frL.i', 'Xitm.fr.o'), ('Xitm.bkL.i', 'Xitm.bk.o'), ('Xetm.bk.i', 'Xetm.fr.o'), ('Xetm.bk.i', 'Xetm.bk.o'), ('Xetm.fr.o', 'Xitm.fr.i'), ('Xetm.fr.i', 'Xetm.bk.o'), ('Xetm.fr.i', 'Xetm.fr.o'), ('Xetm.fr.i', 'Xetm.fr.i.tp'), ('Xetm.frL.i', 'Xetm.fr.o'), ('Xetm.bkL.i', 'Xetm.bk.o'), ('Xetm.fr.o.exc', 'Xetm.fr.o'), ('Yitm.bk.i', 'Yitm.fr.o'), ('Yitm.bk.i', 'Yitm.bk.o'), ('Yitm.bk.i', 'Yitm.bk.i.tp'), ('Yitm.fr.o', 'Yetm.fr.i'), ('Yitm.fr.i', 'Yitm.bk.o'), ('Yitm.fr.i', 'Yitm.fr.o'), ('Yitm.bk.o', 'bs.frB.i'), ('Yitm.frL.i', 'Yitm.fr.o'), ('Yitm.bkL.i', 'Yitm.bk.o'), ('Yetm.bk.i', 'Yetm.fr.o'), ('Yetm.bk.i', 'Yetm.bk.o'), ('Yetm.fr.o', 'Yitm.fr.i'), ('Yetm.fr.i', 'Yetm.bk.o'), ('Yetm.fr.i', 'Yetm.fr.o'), ('Yetm.fr.i', 'Yetm.fr.i.tp'), ('Yetm.frL.i', 'Yetm.fr.o'), ('Yetm.bkL.i', 'Yetm.bk.o'), ('Yetm.fr.o.exc', 'Yetm.fr.o'), ('bs.frA.i', 'bs.bkA.o'), ('bs.frA.i', 'bs.frB.o'), ('bs.bkA.o', 'Xitm.bk.i'), ('bs.bkA.i', 'bs.frA.o'), ('bs.bkA.i', 'bs.bkB.o'), ('bs.frA.o', 'prm.fr.i'), ('bs.frB.i', 'bs.bkB.o'), ('bs.frB.i', 'bs.frA.o'), ('bs.bkB.o', 'opo.M1.frA.i'), ('bs.bkB.i', 'bs.frB.o'), ('bs.bkB.i', 'bs.bkA.o'), ('bs.frB.o', 'Yitm.bk.i'), ('opo.M1.frA.i', 'opo.M1.bkA.o'), ('opo.M1.frA.i', 'opo.M1.frB.o'), ('opo.M1.bkA.o', 'opo.M2.frB.i'), ('opo.M1.bkA.i', 'opo.M1.frA.o'), ('opo.M1.bkA.i', 'opo.M1.bkB.o'), ('opo.M1.frA.o', 'bs.bkB.i'), ('opo.M1.frB.i', 'opo.M1.bkB.o'), ('opo.M1.frB.i', 'opo.M1.frA.o'), ('opo.M1.bkB.o', 'opo.SQZ.fr.i'), ('opo.M1.bkB.i', 'opo.M1.frB.o'), ('opo.M1.bkB.i', 'opo.M1.bkA.o'), ('opo.M1.frB.o', 'srm.fr.i'), ('opo.M2.frA.i', 'opo.M2.bkA.o'), ('opo.M2.frA.i', 'opo.M2.frB.o'), ('opo.M2.bkA.i', 'opo.M2.frA.o'), ('opo.M2.bkA.i', 'opo.M2.bkB.o'), ('opo.M2.frA.o', 'opo.SQZ.bk.i'), ('opo.M2.frB.i', 'opo.M2.bkB.o'), ('opo.M2.frB.i', 'opo.M2.frA.o'), ('opo.M2.bkB.i', 'opo.M2.frB.o'), ('opo.M2.bkB.i', 'opo.M2.bkA.o'), ('opo.M2.frB.o', 'opo.M1.bkA.i'), ('opo.SQZ.bk.i', 'opo.SQZ.fr.o'), ('opo.SQZ.bk.i', 'opo.SQZ.bk.o'), ('opo.SQZ.fr.o', 'opo.M1.bkB.i'), ('opo.SQZ.fr.i', 'opo.SQZ.bk.o'), ('opo.SQZ.fr.i', 'opo.SQZ.fr.o'), ('opo.SQZ.bk.o', 'opo.M2.frA.i'), ('opo.SQZ.frL.i', 'opo.SQZ.fr.o'), ('opo.SQZ.bkL.i', 'opo.SQZ.bk.o'), ('srm.bk.i', 'srm.fr.o'), ('srm.bk.i', 'srm.bk.o'), ('srm.fr.o', 'opo.M1.frB.i'), ('srm.fr.i', 'srm.bk.o'), ('srm.fr.i', 'srm.fr.o'), ('srm.bk.o', 'srm.bk.o.tp'), ('srm.frL.i', 'srm.fr.o'), ('srm.bkL.i', 'srm.bk.o'), ('srm.bk.i.exc', 'srm.bk.i')])
captured errors:
def T_SFLU_OPODRFPMI_build_show():
        """
        Show a graph reduction using networkx+tikz
        """
        # opo['Yetm'].node_angle["fr.o"] = -45
        # opo['Yitm'].node_angle["bk.o"] = -45
        # opo['srm'].node_angle["bk.o"] = -45
        # opo['srm'].edges["bk.i", "bk.i.exc"] = "sec_to"
        # opo['srm'].edges["bk.o.tp", "bk.o"] = "sec_fr"
        # opo['srm'].edge_handedness["bk.o.tp", "bk.o"] = "r"
        opo = build_OPO()

        ifo = optics.GraphElement()
        ifo.subgraph_add(
            'prm', optics.LossyBasisMirror(),
            translation_xy=(-25, 0),
            rotation_deg=180,
        )
        ifo.subgraph_add(
            'Xitm', optics.LossyBasisMirror(),
            translation_xy=(25, 0),
            rotation_deg=180,
        )
        ifo.subgraph_add(
            'Xetm', optics.LossyMirror(),
            translation_xy=(55, 0),
            rotation_deg=0
        )
        ifo.subgraph_add(
            'Yitm', optics.LossyBasisMirror(),
            translation_xy=(0, 25),
            rotation_deg=90+180,
        )
        ifo.subgraph_add(
            'Yetm', optics.LossyMirror(),
            translation_xy=(0, 55),
            rotation_deg=90,
        )
        ifo.subgraph_add(
            'bs', optics.BeamSplitter(),
            translation_xy=(0, 0),
            rotation_deg=0
        )
        ifo.subgraph_add(
            'opo', opo,
            translation_xy=(0, -60),
            rotation_deg=-90
        )
        ifo.subgraph_add(
            'srm', optics.LossyBasisMirror(),
            translation_xy=(+30, -40),
            rotation_deg=0,
        )
        ifo.edges.update({
            ("prm.fr.i",  "bs.frA.o"): "prc.tau",
            ("bs.frA.i",  "prm.fr.o"): "prc.tau",

            ("opo.M1.frA.i", "bs.bkB.o"): "sec.tau",
            ("bs.bkB.i", "opo.M1.frA.o"): "sec.tau",
            ("srm.fr.i", "opo.M1.frB.o"): "sec.tau",
            ("opo.M1.frB.i", "srm.fr.o"): "sec.tau",

            ("Yitm.bk.i", "bs.frB.o"): 'BSY.tau',
            ("bs.frB.i",  "Yitm.bk.o"): 'BSY.tau',

            ("Xitm.bk.i", "bs.bkA.o"): 'BSX.tau',
            ("bs.bkA.i",  "Xitm.bk.o"): 'BSX.tau',

            ("Xetm.fr.i", "Xitm.fr.o"): "XARM.tau",
            ("Xitm.fr.i", "Xetm.fr.o"): "XARM.tau",

            ("Yetm.fr.i", "Yitm.fr.o"): "YARM.tau",
            ("Yitm.fr.i", "Yetm.fr.o"): "YARM.tau",
        })
        ifo['Yetm'].node_angle["fr.o"] = -45
        ifo['Yitm'].node_angle["bk.o"] = -45
        ifo['srm'].node_angle["bk.o"] = -45
        ifo['srm'].edges["bk.i", "bk.i.exc"] = "sec_to"
        ifo['srm'].edges["bk.o.tp", "bk.o"] = "sec_fr"
        ifo['srm'].edge_handedness["bk.o.tp", "bk.o"] = "r"

        ifo['srm'].locations["bk.i.exc"] = (15, -10)
        ifo['srm'].locations["bk.o.tp"] = (15, 10)

        ifo['prm'].edges["bk.i", "bk.i.exc"] = "1"
        ifo['prm'].edges["bk.o.tp", "bk.o"] = "1"
        ifo['prm'].node_angle["bk.i.exc"] = +45

        ifo['prm'].locations["bk.i.exc"] = (15, -10)
        ifo['prm'].locations["bk.o.tp"] = (15, 10)
        ifo['prm'].edges["fr.i.tp", "fr.i"] = "1"
        ifo['prm'].locations["fr.i.tp"] = (-5, 12)

        ifo['Xetm'].edges["fr.i.tp", "fr.i"] = "1"
        ifo['Xetm'].edges["fr.o", "fr.o.exc"] = "1"
        ifo['Xetm'].locations["fr.i.tp"] = (-5, 15)
        ifo['Xetm'].locations["fr.o.exc"] = (-5, -15)

        ifo['Yetm'].edges["fr.i.tp", "fr.i"] = "1"
        ifo['Yetm'].edge_handedness["fr.i.tp", "fr.i"] = "r"
        ifo['Yetm'].edges["fr.o", "fr.o.exc"] = "1"
        ifo['Yetm'].locations["fr.i.tp"] = (-5, 15)
        ifo['Yetm'].locations["fr.o.exc"] = (-5, -15)

        ifo['Xitm'].edges["bk.i.tp", "bk.i"] = "1"
        ifo['Xitm'].locations["bk.i.tp"] = (5, -12)
        ifo['Yitm'].edges["bk.i.tp", "bk.i"] = "1"
        ifo['Yitm'].locations["bk.i.tp"] = (5, -12)

        sflu = SFLU.SFLU(
            edges=ifo.build_edges(),
            derivatives=[
            ],
            reduce_list=reduce_list,
            graph=True,
        )
        # match=False allows a reduced input/output set
        # sflu.graph_nodes_pos(ifo.build_locations(), match=True)
        ifo.update_sflu(sflu)
        G1 = sflu.G.copy()

        if True:
            yamlstr = sflu.convert_self2yamlstr()
            # print(yamlstr)
            with open(tjoin('DRFPMI.yaml'), 'w') as F:
                F.write(yamlstr)
            sflu = SFLU.SFLU.convert_yamlstr2self(yamlstr)
        #sflu.graph_nodes_pos(DRFPMI_locs, match=True)

        print('inputs: ', sflu.inputs)
        print('outputs: ', sflu.outputs)
        print('nodes: ', sflu.nodes)
        # dprint('edges: ', sflu.edges)

        print('nodes')
        #print(sflu.graph_nodes_repr())
        #G1 = sflu.G.copy()
        sflu.graph_reduce_auto_pos(lX=-10, rX=+10, Y=0, dY=-2)
        #sflu.reduce(*reduce_list)
        #assert(not sflu.nodes)
        #sflu.graph_reduce_auto_pos_io(lX=-30, rX=+30, Y=10, dY=-2)
        #G2 = sflu.G.copy()
        dprint(G1.edges)

>       nx2tikz.dump_pdf(
            [
                G1,
                #G2,
            ],
            fname = tjoin('testG.pdf'),
            texname = tjoin('testG.tex'),
            # preamble = preamble,
            scale='10pt',
        )

../../src/wield/control/SFLU/test/T_SFLU_OPODRFPMI.py:264: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
../../src/wield/control/SFLU/nx2tikz.py:230: in dump_pdf
    subprocess.run(opt, check=True, capture_output=True)
/opt/conda/lib/python3.12/subprocess.py:548: in run
    with Popen(*popenargs, **kwargs) as process:
/opt/conda/lib/python3.12/subprocess.py:1026: in __init__
    self._execute_child(args, executable, preexec_fn, close_fds,
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <Popen: returncode: 255 args: ['pdflatex', '-interaction=batchmode', '-outpu...>
args = ['pdflatex', '-interaction=batchmode', '-output-directory=/builds/wield-control/docs/maketests/test_results/T_SFLU_OPO...gvw9gm', '/builds/wield-control/docs/maketests/test_results/T_SFLU_OPODRFPMI.py/T_SFLU_OPODRFPMI_build_show/testG.tex']
executable = b'pdflatex', preexec_fn = None, close_fds = True, pass_fds = ()
cwd = None, env = None, startupinfo = None, creationflags = 0, shell = False
p2cread = -1, p2cwrite = -1, c2pread = 5, c2pwrite = 6, errread = 7
errwrite = 8, restore_signals = True, gid = None, gids = None, uid = None
umask = -1, start_new_session = False, process_group = -1

    def _execute_child(self, args, executable, preexec_fn, close_fds,
                       pass_fds, cwd, env,
                       startupinfo, creationflags, shell,
                       p2cread, p2cwrite,
                       c2pread, c2pwrite,
                       errread, errwrite,
                       restore_signals,
                       gid, gids, uid, umask,
                       start_new_session, process_group):
        """Execute program (POSIX version)"""

        if isinstance(args, (str, bytes)):
            args = [args]
        elif isinstance(args, os.PathLike):
            if shell:
                raise TypeError('path-like args is not allowed when '
                                'shell is true')
            args = [args]
        else:
            args = list(args)

        if shell:
            # On Android the default shell is at '/system/bin/sh'.
            unix_shell = ('/system/bin/sh' if
                      hasattr(sys, 'getandroidapilevel') else '/bin/sh')
            args = [unix_shell, "-c"] + args
            if executable:
                args[0] = executable

        if executable is None:
            executable = args[0]

        sys.audit("subprocess.Popen", executable, args, cwd, env)

        if (_USE_POSIX_SPAWN
                and os.path.dirname(executable)
                and preexec_fn is None
                and not close_fds
                and not pass_fds
                and cwd is None
                and (p2cread == -1 or p2cread > 2)
                and (c2pwrite == -1 or c2pwrite > 2)
                and (errwrite == -1 or errwrite > 2)
                and not start_new_session
                and process_group == -1
                and gid is None
                and gids is None
                and uid is None
                and umask < 0):
            self._posix_spawn(args, executable, env, restore_signals,
                              p2cread, p2cwrite,
                              c2pread, c2pwrite,
                              errread, errwrite)
            return

        orig_executable = executable

        # For transferring possible exec failure from child to parent.
        # Data format: "exception name:hex errno:description"
        # Pickle is not used; it is complex and involves memory allocation.
        errpipe_read, errpipe_write = os.pipe()
        # errpipe_write must not be in the standard io 0, 1, or 2 fd range.
        low_fds_to_close = []
        while errpipe_write < 3:
            low_fds_to_close.append(errpipe_write)
            errpipe_write = os.dup(errpipe_write)
        for low_fd in low_fds_to_close:
            os.close(low_fd)
        try:
            try:
                # We must avoid complex work that could involve
                # malloc or free in the child process to avoid
                # potential deadlocks, thus we do all this here.
                # and pass it to fork_exec()

                if env is not None:
                    env_list = []
                    for k, v in env.items():
                        k = os.fsencode(k)
                        if b'=' in k:
                            raise ValueError("illegal environment variable name")
                        env_list.append(k + b'=' + os.fsencode(v))
                else:
                    env_list = None  # Use execv instead of execve.
                executable = os.fsencode(executable)
                if os.path.dirname(executable):
                    executable_list = (executable,)
                else:
                    # This matches the behavior of os._execvpe().
                    executable_list = tuple(
                        os.path.join(os.fsencode(dir), executable)
                        for dir in os.get_exec_path(env))
                fds_to_keep = set(pass_fds)
                fds_to_keep.add(errpipe_write)
                self.pid = _fork_exec(
                        args, executable_list,
                        close_fds, tuple(sorted(map(int, fds_to_keep))),
                        cwd, env_list,
                        p2cread, p2cwrite, c2pread, c2pwrite,
                        errread, errwrite,
                        errpipe_read, errpipe_write,
                        restore_signals, start_new_session,
                        process_group, gid, gids, uid, umask,
                        preexec_fn, _USE_VFORK)
                self._child_created = True
            finally:
                # be sure the FD is closed no matter what
                os.close(errpipe_write)

            self._close_pipe_fds(p2cread, p2cwrite,
                                 c2pread, c2pwrite,
                                 errread, errwrite)

            # Wait for exec to fail or succeed; possibly raising an
            # exception (limited in size)
            errpipe_data = bytearray()
            while True:
                part = os.read(errpipe_read, 50000)
                errpipe_data += part
                if not part or len(errpipe_data) > 50000:
                    break
        finally:
            # be sure the FD is closed no matter what
            os.close(errpipe_read)

        if errpipe_data:
            try:
                pid, sts = os.waitpid(self.pid, 0)
                if pid == self.pid:
                    self._handle_exitstatus(sts)
                else:
                    self.returncode = sys.maxsize
            except ChildProcessError:
                pass

            try:
                exception_name, hex_errno, err_msg = (
                        errpipe_data.split(b':', 2))
                # The encoding here should match the encoding
                # written in by the subprocess implementations
                # like _posixsubprocess
                err_msg = err_msg.decode()
            except ValueError:
                exception_name = b'SubprocessError'
                hex_errno = b'0'
                err_msg = 'Bad exception data from child: {!r}'.format(
                              bytes(errpipe_data))
            child_exception_type = getattr(
                    builtins, exception_name.decode('ascii'),
                    SubprocessError)
            if issubclass(child_exception_type, OSError) and hex_errno:
                errno_num = int(hex_errno, 16)
                if err_msg == "noexec:chdir":
                    err_msg = ""
                    # The error must be from chdir(cwd).
                    err_filename = cwd
                elif err_msg == "noexec":
                    err_msg = ""
                    err_filename = None
                else:
                    err_filename = orig_executable
                if errno_num != 0:
                    err_msg = os.strerror(errno_num)
                if err_filename is not None:
>                   raise child_exception_type(errno_num, err_msg, err_filename)
E                   FileNotFoundError: [Errno 2] No such file or directory: 'pdflatex'

/opt/conda/lib/python3.12/subprocess.py:1955: FileNotFoundError
T_SFLU_OPO_build_show()[source][github]

Show a graph reduction using networkx+tikz

code
docstring
"""
Show a graph reduction using networkx+tikz
"""

 1def T_SFLU_OPO_build_show():
 2
 3    # opo['Yetm'].node_angle["fr.o"] = -45
 4    # opo['Yitm'].node_angle["bk.o"] = -45
 5    # opo['srm'].node_angle["bk.o"] = -45
 6    # opo['srm'].edges["bk.i", "bk.i.exc"] = "sec_to"
 7    # opo['srm'].edges["bk.o.tp", "bk.o"] = "sec_fr"
 8    # opo['srm'].edge_handedness["bk.o.tp", "bk.o"] = "r"
 9    opo = build_OPO()
10
11    sflu = SFLU.SFLU(
12        edges=opo.build_edges(),
13        derivatives=[
14        ],
15        reduce_list=reduce_list,
16        graph=True,
17    )
18    # match=False allows a reduced input/output set
19    # sflu.graph_nodes_pos(ifo.build_locations(), match=True)
20    opo.update_sflu(sflu)
21    G1 = sflu.G.copy()
22
23    if True:
24        yamlstr = sflu.convert_self2yamlstr()
25        # print(yamlstr)
26        with open(tjoin('DRFPMI.yaml'), 'w') as F:
27            F.write(yamlstr)
28        sflu = SFLU.SFLU.convert_yamlstr2self(yamlstr)
29    #sflu.graph_nodes_pos(DRFPMI_locs, match=True)
30
31    print('inputs: ', sflu.inputs)
32    print('outputs: ', sflu.outputs)
33    print('nodes: ', sflu.nodes)
34    # dprint('edges: ', sflu.edges)
35
36    print('nodes')
37    #print(sflu.graph_nodes_repr())
38    #G1 = sflu.G.copy()
39    sflu.graph_reduce_auto_pos(lX=-10, rX=+10, Y=0, dY=-2)
40    #sflu.reduce(*reduce_list)
41    #assert(not sflu.nodes)
42    #sflu.graph_reduce_auto_pos_io(lX=-30, rX=+30, Y=10, dY=-2)
43    #G2 = sflu.G.copy()
44    dprint(G1.edges)
45
46    nx2tikz.dump_pdf(
47        [
48            G1,
49            #G2,
50        ],
51        fname = tjoin('testG.pdf'),
52        texname = tjoin('testG.tex'),
53        # preamble = preamble,
54        scale='10pt',
55    )
pytest information

This code is wrapped in a pytest function using conventions detailed in Pytest Conventions. The full name of this test, as known by the documentation, is:

wield.control.SFLU.test.T_SFLU_OPODRFPMI.T_SFLU_OPO_build_show

The full name is useful when building documentation, to link a reference to this page using :func:`name`, or directly include it with an autofunction directive. The collapse nodes below show every instance of the test run. There may only be one, but if the test was run multiple times through pytest parametrizations, the list can be longer.

T_SFLU_OPO_build_show
output
inputs:  {'SQZ.bkL.i', 'M1.frA.i', 'M2.bkA.i', 'SQZ.frL.i', 'M1.frB.i', 'M2.bkB.i'}
outputs:  {'M1.frA.o', 'M2.bkB.o', 'M1.frB.o', 'M2.bkA.o'}
nodes:  {'M1.bkA.o', 'M2.frB.o', 'M1.bkB.o', 'SQZ.fr.o', 'M1.bkA.i', 'SQZ.bk.o', 'SQZ.bk.i', 'M2.frA.i', 'M1.bkB.i', 'SQZ.fr.i', 'M2.frB.i', 'M2.frA.o'}
nodes
Captured stderr call
OutEdgeView([('M1.frA.i', 'M1.bkA.o'), ('M1.frA.i', 'M1.frB.o'), ('M1.bkA.o', 'M2.frB.i'), ('M1.bkA.i', 'M1.frA.o'), ('M1.bkA.i', 'M1.bkB.o'), ('M1.frB.i', 'M1.bkB.o'), ('M1.frB.i', 'M1.frA.o'), ('M1.bkB.o', 'SQZ.fr.i'), ('M1.bkB.i', 'M1.frB.o'), ('M1.bkB.i', 'M1.bkA.o'), ('M2.frA.i', 'M2.bkA.o'), ('M2.frA.i', 'M2.frB.o'), ('M2.bkA.i', 'M2.frA.o'), ('M2.bkA.i', 'M2.bkB.o'), ('M2.frA.o', 'SQZ.bk.i'), ('M2.frB.i', 'M2.bkB.o'), ('M2.frB.i', 'M2.frA.o'), ('M2.bkB.i', 'M2.frB.o'), ('M2.bkB.i', 'M2.bkA.o'), ('M2.frB.o', 'M1.bkA.i'), ('SQZ.bk.i', 'SQZ.fr.o'), ('SQZ.bk.i', 'SQZ.bk.o'), ('SQZ.fr.o', 'M1.bkB.i'), ('SQZ.fr.i', 'SQZ.bk.o'), ('SQZ.fr.i', 'SQZ.fr.o'), ('SQZ.bk.o', 'M2.frA.i'), ('SQZ.frL.i', 'SQZ.fr.o'), ('SQZ.bkL.i', 'SQZ.bk.o')])
captured errors:
def T_SFLU_OPO_build_show():
        """
        Show a graph reduction using networkx+tikz
        """
        # opo['Yetm'].node_angle["fr.o"] = -45
        # opo['Yitm'].node_angle["bk.o"] = -45
        # opo['srm'].node_angle["bk.o"] = -45
        # opo['srm'].edges["bk.i", "bk.i.exc"] = "sec_to"
        # opo['srm'].edges["bk.o.tp", "bk.o"] = "sec_fr"
        # opo['srm'].edge_handedness["bk.o.tp", "bk.o"] = "r"
        opo = build_OPO()

        sflu = SFLU.SFLU(
            edges=opo.build_edges(),
            derivatives=[
            ],
            reduce_list=reduce_list,
            graph=True,
        )
        # match=False allows a reduced input/output set
        # sflu.graph_nodes_pos(ifo.build_locations(), match=True)
        opo.update_sflu(sflu)
        G1 = sflu.G.copy()

        if True:
            yamlstr = sflu.convert_self2yamlstr()
            # print(yamlstr)
            with open(tjoin('DRFPMI.yaml'), 'w') as F:
                F.write(yamlstr)
            sflu = SFLU.SFLU.convert_yamlstr2self(yamlstr)
        #sflu.graph_nodes_pos(DRFPMI_locs, match=True)

        print('inputs: ', sflu.inputs)
        print('outputs: ', sflu.outputs)
        print('nodes: ', sflu.nodes)
        # dprint('edges: ', sflu.edges)

        print('nodes')
        #print(sflu.graph_nodes_repr())
        #G1 = sflu.G.copy()
        sflu.graph_reduce_auto_pos(lX=-10, rX=+10, Y=0, dY=-2)
        #sflu.reduce(*reduce_list)
        #assert(not sflu.nodes)
        #sflu.graph_reduce_auto_pos_io(lX=-30, rX=+30, Y=10, dY=-2)
        #G2 = sflu.G.copy()
        dprint(G1.edges)

>       nx2tikz.dump_pdf(
            [
                G1,
                #G2,
            ],
            fname = tjoin('testG.pdf'),
            texname = tjoin('testG.tex'),
            # preamble = preamble,
            scale='10pt',
        )

../../src/wield/control/SFLU/test/T_SFLU_OPODRFPMI.py:108: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
../../src/wield/control/SFLU/nx2tikz.py:230: in dump_pdf
    subprocess.run(opt, check=True, capture_output=True)
/opt/conda/lib/python3.12/subprocess.py:548: in run
    with Popen(*popenargs, **kwargs) as process:
/opt/conda/lib/python3.12/subprocess.py:1026: in __init__
    self._execute_child(args, executable, preexec_fn, close_fds,
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <Popen: returncode: 255 args: ['pdflatex', '-interaction=batchmode', '-outpu...>
args = ['pdflatex', '-interaction=batchmode', '-output-directory=/builds/wield-control/docs/maketests/test_results/T_SFLU_OPO.../tmpqosyzs54', '/builds/wield-control/docs/maketests/test_results/T_SFLU_OPODRFPMI.py/T_SFLU_OPO_build_show/testG.tex']
executable = b'pdflatex', preexec_fn = None, close_fds = True, pass_fds = ()
cwd = None, env = None, startupinfo = None, creationflags = 0, shell = False
p2cread = -1, p2cwrite = -1, c2pread = 5, c2pwrite = 6, errread = 7
errwrite = 8, restore_signals = True, gid = None, gids = None, uid = None
umask = -1, start_new_session = False, process_group = -1

    def _execute_child(self, args, executable, preexec_fn, close_fds,
                       pass_fds, cwd, env,
                       startupinfo, creationflags, shell,
                       p2cread, p2cwrite,
                       c2pread, c2pwrite,
                       errread, errwrite,
                       restore_signals,
                       gid, gids, uid, umask,
                       start_new_session, process_group):
        """Execute program (POSIX version)"""

        if isinstance(args, (str, bytes)):
            args = [args]
        elif isinstance(args, os.PathLike):
            if shell:
                raise TypeError('path-like args is not allowed when '
                                'shell is true')
            args = [args]
        else:
            args = list(args)

        if shell:
            # On Android the default shell is at '/system/bin/sh'.
            unix_shell = ('/system/bin/sh' if
                      hasattr(sys, 'getandroidapilevel') else '/bin/sh')
            args = [unix_shell, "-c"] + args
            if executable:
                args[0] = executable

        if executable is None:
            executable = args[0]

        sys.audit("subprocess.Popen", executable, args, cwd, env)

        if (_USE_POSIX_SPAWN
                and os.path.dirname(executable)
                and preexec_fn is None
                and not close_fds
                and not pass_fds
                and cwd is None
                and (p2cread == -1 or p2cread > 2)
                and (c2pwrite == -1 or c2pwrite > 2)
                and (errwrite == -1 or errwrite > 2)
                and not start_new_session
                and process_group == -1
                and gid is None
                and gids is None
                and uid is None
                and umask < 0):
            self._posix_spawn(args, executable, env, restore_signals,
                              p2cread, p2cwrite,
                              c2pread, c2pwrite,
                              errread, errwrite)
            return

        orig_executable = executable

        # For transferring possible exec failure from child to parent.
        # Data format: "exception name:hex errno:description"
        # Pickle is not used; it is complex and involves memory allocation.
        errpipe_read, errpipe_write = os.pipe()
        # errpipe_write must not be in the standard io 0, 1, or 2 fd range.
        low_fds_to_close = []
        while errpipe_write < 3:
            low_fds_to_close.append(errpipe_write)
            errpipe_write = os.dup(errpipe_write)
        for low_fd in low_fds_to_close:
            os.close(low_fd)
        try:
            try:
                # We must avoid complex work that could involve
                # malloc or free in the child process to avoid
                # potential deadlocks, thus we do all this here.
                # and pass it to fork_exec()

                if env is not None:
                    env_list = []
                    for k, v in env.items():
                        k = os.fsencode(k)
                        if b'=' in k:
                            raise ValueError("illegal environment variable name")
                        env_list.append(k + b'=' + os.fsencode(v))
                else:
                    env_list = None  # Use execv instead of execve.
                executable = os.fsencode(executable)
                if os.path.dirname(executable):
                    executable_list = (executable,)
                else:
                    # This matches the behavior of os._execvpe().
                    executable_list = tuple(
                        os.path.join(os.fsencode(dir), executable)
                        for dir in os.get_exec_path(env))
                fds_to_keep = set(pass_fds)
                fds_to_keep.add(errpipe_write)
                self.pid = _fork_exec(
                        args, executable_list,
                        close_fds, tuple(sorted(map(int, fds_to_keep))),
                        cwd, env_list,
                        p2cread, p2cwrite, c2pread, c2pwrite,
                        errread, errwrite,
                        errpipe_read, errpipe_write,
                        restore_signals, start_new_session,
                        process_group, gid, gids, uid, umask,
                        preexec_fn, _USE_VFORK)
                self._child_created = True
            finally:
                # be sure the FD is closed no matter what
                os.close(errpipe_write)

            self._close_pipe_fds(p2cread, p2cwrite,
                                 c2pread, c2pwrite,
                                 errread, errwrite)

            # Wait for exec to fail or succeed; possibly raising an
            # exception (limited in size)
            errpipe_data = bytearray()
            while True:
                part = os.read(errpipe_read, 50000)
                errpipe_data += part
                if not part or len(errpipe_data) > 50000:
                    break
        finally:
            # be sure the FD is closed no matter what
            os.close(errpipe_read)

        if errpipe_data:
            try:
                pid, sts = os.waitpid(self.pid, 0)
                if pid == self.pid:
                    self._handle_exitstatus(sts)
                else:
                    self.returncode = sys.maxsize
            except ChildProcessError:
                pass

            try:
                exception_name, hex_errno, err_msg = (
                        errpipe_data.split(b':', 2))
                # The encoding here should match the encoding
                # written in by the subprocess implementations
                # like _posixsubprocess
                err_msg = err_msg.decode()
            except ValueError:
                exception_name = b'SubprocessError'
                hex_errno = b'0'
                err_msg = 'Bad exception data from child: {!r}'.format(
                              bytes(errpipe_data))
            child_exception_type = getattr(
                    builtins, exception_name.decode('ascii'),
                    SubprocessError)
            if issubclass(child_exception_type, OSError) and hex_errno:
                errno_num = int(hex_errno, 16)
                if err_msg == "noexec:chdir":
                    err_msg = ""
                    # The error must be from chdir(cwd).
                    err_filename = cwd
                elif err_msg == "noexec":
                    err_msg = ""
                    err_filename = None
                else:
                    err_filename = orig_executable
                if errno_num != 0:
                    err_msg = os.strerror(errno_num)
                if err_filename is not None:
>                   raise child_exception_type(errno_num, err_msg, err_filename)
E                   FileNotFoundError: [Errno 2] No such file or directory: 'pdflatex'

/opt/conda/lib/python3.12/subprocess.py:1955: FileNotFoundError
build_OPO()[source][github]

This is a pytest needing documentation

code
 1def build_OPO():
 2    opo = optics.GraphElement()
 3    opo.subgraph_add(
 4        'M1', optics.BeamSplitter(),
 5        translation_xy=(-20, 0),
 6        rotation_deg=0
 7    )
 8    opo.subgraph_add(
 9        'M2', optics.BeamSplitter(),
10        translation_xy=(20, 0),
11        rotation_deg=90
12    )
13    opo.subgraph_add(
14        'SQZ', optics.LossyBasisMirror(),
15        translation_xy=(0, -25),
16        rotation_deg=0
17    )
18    opo.edges.update({
19        ("SQZ.fr.i",  "M1.bkB.o"): "1",
20        ("M1.bkB.i",  "SQZ.fr.o"): "1",
21        ("SQZ.bk.i",  "M2.frA.o"): "1",
22        ("M2.frA.i",  "SQZ.bk.o"): "1",
23        ("M1.bkA.i",  "M2.frB.o"): "1",
24        ("M2.frB.i",  "M1.bkA.o"): "1",
25    })
26    return opo
pytest information

This code is wrapped in a pytest function using conventions detailed in Pytest Conventions. The full name of this test, as known by the documentation, is:

wield.control.SFLU.test.T_SFLU_OPODRFPMI.build_OPO

The full name is useful when building documentation, to link a reference to this page using :func:`name`, or directly include it with an autofunction directive. The collapse nodes below show every instance of the test run. There may only be one, but if the test was run multiple times through pytest parametrizations, the list can be longer.