-
Notifications
You must be signed in to change notification settings - Fork 184
/
klive.lym
133 lines (107 loc) · 3.99 KB
/
klive.lym
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
<?xml version="1.0" encoding="utf-8"?>
<klayout-macro>
<description>klive</description>
<version/>
<category>pymacros</category>
<prolog/>
<epilog/>
<doc/>
<autorun>true</autorun>
<autorun-early>false</autorun-early>
<shortcut/>
<show-in-menu>false</show-in-menu>
<group-name/>
<menu-path>gdsfactory.begin</menu-path>
<interpreter>python</interpreter>
<dsl-interpreter-name/>
<text>from __future__ import print_function
import pya
import re
import json
import sys
import tempfile
import os
import pathlib
klayout_version = pya.Application.instance().version()
is_python3 = sys.version_info.major==3
temp = tempfile.TemporaryDirectory().name
if is_python3:
import codecs
class MyServer(pya.QTcpServer):
"""
Implements a TCP server listening on port 8082.
You can use it to instantly load a GDS file, programmatically, from Python.
Just send a JSON-formatted command to localhost:8082.
See README for more details.
"""
def new_connection(self):
""" Handler for a new connection
"""
try:
url = None
# Get a new connection object
connection = self.nextPendingConnection()
# Read in the request
data = None
while connection.isOpen() and connection.state() == pya.QTcpSocket.ConnectedState:
if connection.canReadLine():
line = connection.readLine()
if is_python3:
pass
else:
line = line.decode("utf-8")
data = json.loads(line)
else:
connection.waitForReadyRead(100)
# Interpret the data
gds_path = data["gds"]
# Store the current view
window = pya.Application.instance().main_window()
current_view = window.current_view()
previous_view = current_view.box() if current_view else None
# temp = pathlib.Path(tempfile.TemporaryDirectory().name)
# temp.mkdir(exist_ok=True)
# filepath_layers = temp / 'klayout_layers.lyp'
filepath_layers = os.path.join(temp, 'klayout_layers.lyp')
os.makedirs(temp, exist_ok=True)
if current_view:
current_view.save_layer_props(str(filepath_layers))
# Load the new layout
window.load_layout(gds_path, 0)
if current_view and os.path.exists(filepath_layers):
current_view.load_layer_props(str(filepath_layers), True)
# Restore the previous position
view = window.current_view()
view.max_hier()
if previous_view and data["keep_position"]==True:
view.zoom_box(previous_view)
# Report progress
print("Loaded {}".format(gds_path))
connection.write("Loaded {}".format(gds_path))
# Ping the necessary event
if hasattr(window, "on_klive_update"):
print (window.on_klive_update) # To avoid triggering a false error
if window.on_klive_update:
window.on_klive_update()
# Disconnect
connection.disconnectFromHost()
signal = pya.qt_signal("disconnected()")
slot = pya.qt_slot("deleteLater()")
pya.QObject.connect(connection, signal, connection, slot)
except Exception as ex:
print("ERROR " + str(ex))
def __init__(self, parent=None):
"""
Initialize the server and put into listen mode
"""
super(MyServer, self).__init__(parent)
ha = pya.QHostAddress.new_ip4(0)
self.listen(ha, 8082)
# install an event on a new connection
self.newConnection(self.new_connection)
# Start the server
server = MyServer()
path = pathlib.Path(__file__).parent.absolute()
print("klive 6.16.1 is running from {} inside klayout {} with python {}".format(str(path), klayout_version, sys.version))
</text>
</klayout-macro>