20
20
21
21
22
22
class MPRemote2 :
23
- def __init__ (self , shell : InteractiveShell , port : str = "auto" , resume : bool = True ,):
23
+ def __init__ (
24
+ self ,
25
+ shell : InteractiveShell ,
26
+ port : str = "auto" ,
27
+ resume : bool = True ,
28
+ ):
24
29
self .shell : InteractiveShell = shell
25
30
self .port : str = port # by default connect to the first device
26
31
self .resume = resume # by default resume the device to maintain state
@@ -53,7 +58,9 @@ def run_cmd(
53
58
else :
54
59
log .warning (f"cmd is not a string: { cmd } " )
55
60
log .debug (cmd )
56
- return ipython_run (cmd , stream_out = stream_out , shell = shell , timeout = timeout or self .timeout , follow = follow )
61
+ return ipython_run (
62
+ cmd , stream_out = stream_out , shell = shell , timeout = timeout or self .timeout , follow = follow
63
+ )
57
64
58
65
# output = self.shell.getoutput(cmd, split=True)
59
66
# assert isinstance(output, SList)
@@ -75,23 +82,62 @@ def select_device(self, port: Optional[str], verify: bool = False):
75
82
output = e
76
83
return output
77
84
78
- def run_cell (self , cell : str , * , timeout : Union [int , float ] = TIMEOUT , follow : bool = True ):
85
+ def run_cell (
86
+ self ,
87
+ cell : str ,
88
+ * ,
89
+ timeout : Union [int , float ] = TIMEOUT ,
90
+ follow : bool = True ,
91
+ mount : Optional [str ] = None ,
92
+ ):
79
93
"""run a codeblock on the device and return the output"""
80
- # # TODO: if the cell is small enough, concat the cell with \n and use exec instead of copy
81
- # # - may need escaping quotes and newlines
82
- # copy the cell to a file on the device
83
- self .cell_to_mcu_file (cell , "__magic.py" )
84
- # run the transferred cell/file
85
- result = self .run_mcu_file ("__magic.py" , stream_out = True , timeout = timeout , follow = follow )
86
- return result # ?
87
-
88
- def run_mcu_file (self , filename : str , stream_out : bool = True , timeout : Union [int , float ] = 0 , follow : bool = True ):
89
- exec_cmd = f"exec \" exec( open('{ filename } ').read() , globals() )\" "
94
+ """copy cell to a file and run it on the MCU"""
95
+ with tempfile .NamedTemporaryFile (mode = "w" , suffix = ".py" , delete = False ) as f :
96
+ f .write (
97
+ "# Jupyter cell\n "
98
+ ) # add a line to replace the cell magic to keep the line numbers aligned
99
+ f .write (cell )
100
+ f .close ()
101
+ # copy the file to the device
102
+ run_cmd = f"run { f .name } "
103
+ if mount :
104
+ # prefix the run command with a mount command
105
+ run_cmd = f'mount "{ Path (mount ).as_posix ()} " ' + run_cmd
106
+
107
+ # TODO: detect / retry / report errors copying the file
108
+ result = self .run_cmd (
109
+ run_cmd ,
110
+ stream_out = True ,
111
+ timeout = timeout ,
112
+ follow = follow ,
113
+ )
114
+ # log.info(_)
115
+ # log.info(f.name, "copied to device")
116
+ Path (f .name ).unlink ()
117
+ return result
118
+
119
+ def run_mcu_file (
120
+ self ,
121
+ filename : str ,
122
+ * ,
123
+ stream_out : bool = True ,
124
+ timeout : Union [int , float ] = 0 ,
125
+ follow : bool = True ,
126
+ mount : Optional [str ] = None ,
127
+ ):
128
+ """run a file on the device and return the output"""
129
+ exec_cmd = ""
130
+ if mount :
131
+ exec_cmd = f'mount "{ mount } " '
132
+ exec_cmd += f"exec \" exec( open('{ filename } ').read() , globals() )\" "
90
133
return self .run_cmd (exec_cmd , stream_out = stream_out , timeout = timeout , follow = follow )
91
134
92
- def cell_to_mcu_file (self , cell , filename ):
135
+ def copy_cell_to_mcu (self , cell , * , filename : str ):
136
+ """copy cell to a file to the MCU"""
93
137
with tempfile .NamedTemporaryFile (mode = "w" , suffix = ".py" , delete = False ) as f :
94
- f .write ("# Jupyter cell\n " ) # add a line to replace the cell magic to keep the line numbers aligned
138
+ f .write (
139
+ "# Jupyter cell\n "
140
+ ) # add a line to replace the cell magic to keep the line numbers aligned
95
141
f .write (cell )
96
142
f .close ()
97
143
# copy the file to the device
0 commit comments