%matplotlib inline
import pandas as pd
import socket
host = socket.getfqdn()

from core import  load, zoom, calc, save,plots,monitor

#reload funcs after updating ./core/*.py
import importlib
importlib.reload(load)
importlib.reload(zoom)
importlib.reload(calc)
importlib.reload(save)
importlib.reload(plots)
importlib.reload(monitor)

<module 'core.monitor' from '/ccc/work/cont003/gen7420/odakatin/monitor-sedna/notebook/core/monitor.py'>

# 'month':  = 'JOBID' almost month but not really, 

# If you submit the job with job scheduler, above

#below are list of enviroment variable one can pass
#%env local='2"
# local : if True  run dask local cluster, if not true, put number of workers
# setted in the 'local'
# if no 'local ' given, local will be setted automatically to 'True'
#%env ychunk='2'
#%env tchunk='2'
# controls chunk. 'False' sets no modification from original netcdf file's chunk.  
# ychunk=10 will group the original netcdf file to 10 by 10 
# tchunk=1 will chunk the time coordinate one by one
#%env control=FWC_SSH 
# name of control file to be used for computation/plots/save/ 
#%env file_exp= 
# 'file_exp': Which 'experiment' name is it? 
#.    this corresopnds to intake catalog name without path and .yaml
#%env year=
# for Validation, this correspoinds to path/year/month 's year
# for monitoring, this corresponids to 'date'  having * means do all files in the monitoring directory
# setting it as *0[0-9] &*1[0-9]& *[2-3][0-9], the job can be separated in three lots.
#%env month=
# for monitoring  this corresponds to file path path-XIOS.{month}/
#
#%env save=   proceed saving?   True or False  , Default is setted as True 
#%env plot=   proceed plotting?  True or False , Default is setted as True 
#%env calc=   proceed computation? or just load computed result? True or False , Default is setted as True 
#%env save=False

%%time
# 'savefig': Do we save output in html? or not. keep it true. 
savefig=True
client,cluster,control,catalog_url,month,year,daskreport,outputpath = load.set_control(host)
!mkdir -p $outputpath
!mkdir -p $daskreport
client

local True
using host= irene5791.c-irene.mg1.tgcc.ccc.cea.fr starting dask cluster on local= True workers 16
10000000000
False
not local in tgcc
rome local cluster starting
This code is running on  irene5791.c-irene.mg1.tgcc.ccc.cea.fr using  SEDNA_DELTA_MONITOR file experiment, read from  ../lib/SEDNA_DELTA_MONITOR.yaml  on year= 2012  on month= 01  outputpath= ../results/SEDNA_DELTA_MONITOR/ daskreport= ../results/dask/6312019irene5791.c-irene.mg1.tgcc.ccc.cea.fr_SEDNA_DELTA_MONITOR_01M_Sectiony/
CPU times: user 372 ms, sys: 265 ms, total: 637 ms
Wall time: 11.6 s

Client Scheduler: tcp://127.0.0.1:44546 Dashboard: http://127.0.0.1:8787/status

Cluster Workers: 16 Cores: 256 Memory: 269.57 GB

read plotting information from a csv file

df=load.controlfile(control)
#Take out 'later' tagged computations
#df=df[~df['Value'].str.contains('later')]
df

	Value	Inputs	Equation	Zone	Plot	Colourmap	MinMax	Unit	Oldname	Unnamed: 10
	Section	gridT.votemper,gridS.vosaline,gridV.vomecrty,p...	data.unify_chunks().persist()	FramS	section	None	{'vosaline': (33,36.2), 'votemper': (-2,6), 'v...	None	S-1

Computation starts here

Each computation consists of

1. Load NEMO data set
2. Zoom data set
3. Compute (or load computed data set)
4. Save
5. Plot
6. Close

%%time
import os
calcswitch=os.environ.get('calc', 'True') 
loaddata=((df.Inputs != '').any()) 
print('calcswitch=',calcswitch,'df.Inputs != nothing',loaddata)
data = load.datas(catalog_url,df.Inputs,month,year,daskreport) if ((calcswitch=='True' )*loaddata) else 0 
data

calcswitch= True df.Inputs != nothing True
../lib/SEDNA_DELTA_MONITOR.yaml
using param_xios reading  ../lib/SEDNA_DELTA_MONITOR.yaml
using param_xios reading  <bound method DataSourceBase.describe of sources:
  param_xios:
    args:
      combine: by_coords
      concat_dim: y
      urlpath: /ccc/work/cont003/gen7420/odakatin/CONFIGS/SEDNA/SEDNA-I/SEDNA_Domain_cfg_Tgt_20210423_tsh10m_L1/param_f32/x_*.nc
      xarray_kwargs:
        compat: override
        coords: minimal
        data_vars: minimal
        parallel: true
    description: SEDNA NEMO parameters from MPI output  nav_lon lat fails
    driver: intake_xarray.netcdf.NetCDFSource
    metadata:
      catalog_dir: /ccc/work/cont003/gen7420/odakatin/monitor-sedna/notebook/../lib/
>
{'name': 'param_xios', 'container': 'xarray', 'plugin': ['netcdf'], 'driver': ['netcdf'], 'description': 'SEDNA NEMO parameters from MPI output  nav_lon lat fails', 'direct_access': 'forbid', 'user_parameters': [{'name': 'path', 'description': 'file coordinate', 'type': 'str', 'default': '/ccc/work/cont003/gen7420/odakatin/CONFIGS/SEDNA/MESH/SEDNA_mesh_mask_Tgt_20210423_tsh10m_L1/param'}], 'metadata': {}, 'args': {'urlpath': '/ccc/work/cont003/gen7420/odakatin/CONFIGS/SEDNA/SEDNA-I/SEDNA_Domain_cfg_Tgt_20210423_tsh10m_L1/param_f32/x_*.nc', 'combine': 'by_coords', 'concat_dim': 'y'}}
0 read gridS ['vosaline']
using load_data_xios reading  gridS
using load_data_xios reading  <bound method DataSourceBase.describe of sources:
  data_xios:
    args:
      combine: by_coords
      concat_dim: time_counter,x,y
      urlpath: /ccc/scratch/cont003/gen7420/talandel/SEDNA/SEDNA-DELTA-S/SPLIT/1d/2012/01/*gridS_*_0[0-5][0-9][0-9].nc
      xarray_kwargs:
        compat: override
        coords: minimal
        data_vars: minimal
        drop_variables: !!set
          deptht_bounds: null
          depthu_bounds: null
          nav_lat: null
          nav_lon: null
          time_centerd: null
          time_centered_bounds: null
          time_counter_bounds: null
        parallel: true
        preprocess: !!python/name:core.load.prep ''
    description: SEDNA NEMO outputs from different xios server
    driver: intake_xarray.netcdf.NetCDFSource
    metadata:
      catalog_dir: /ccc/work/cont003/gen7420/odakatin/monitor-sedna/notebook/../lib/
>
      took 54.064802408218384 seconds
0 merging gridS ['vosaline']
1 read gridT ['votemper']
using load_data_xios reading  gridT
using load_data_xios reading  <bound method DataSourceBase.describe of sources:
  data_xios:
    args:
      combine: by_coords
      concat_dim: time_counter,x,y
      urlpath: /ccc/scratch/cont003/gen7420/talandel/SEDNA/SEDNA-DELTA-S/SPLIT/1d/2012/01/*gridT_*_0[0-5][0-9][0-9].nc
      xarray_kwargs:
        compat: override
        coords: minimal
        data_vars: minimal
        drop_variables: !!set
          deptht_bounds: null
          depthu_bounds: null
          nav_lat: null
          nav_lon: null
          time_centerd: null
          time_centered_bounds: null
          time_counter_bounds: null
        parallel: true
        preprocess: !!python/name:core.load.prep ''
    description: SEDNA NEMO outputs from different xios server
    driver: intake_xarray.netcdf.NetCDFSource
    metadata:
      catalog_dir: /ccc/work/cont003/gen7420/odakatin/monitor-sedna/notebook/../lib/
>
      took 4.158769130706787 seconds
1 merging gridT ['votemper']
      took 0.23752760887145996 seconds
2 read gridV ['vomecrty']
using load_data_xios reading  gridV
using load_data_xios reading  <bound method DataSourceBase.describe of sources:
  data_xios:
    args:
      combine: by_coords
      concat_dim: time_counter,x,y
      urlpath: /ccc/scratch/cont003/gen7420/talandel/SEDNA/SEDNA-DELTA-S/SPLIT/1d/2012/01/*gridV_*_0[0-5][0-9][0-9].nc
      xarray_kwargs:
        compat: override
        coords: minimal
        data_vars: minimal
        drop_variables: !!set
          deptht_bounds: null
          depthu_bounds: null
          nav_lat: null
          nav_lon: null
          time_centerd: null
          time_centered_bounds: null
          time_counter_bounds: null
        parallel: true
        preprocess: !!python/name:core.load.prep ''
    description: SEDNA NEMO outputs from different xios server
    driver: intake_xarray.netcdf.NetCDFSource
    metadata:
      catalog_dir: /ccc/work/cont003/gen7420/odakatin/monitor-sedna/notebook/../lib/
>
      took 5.601640462875366 seconds
2 merging gridV ['vomecrty']
      took 0.22144317626953125 seconds
param mask2d will be included in data
param nav_lon will be included in data
param depth will be included in data
param mask will be included in data
param nav_lat will be included in data
sum_num (13, 13, 13, 13, 13, 13, 13, 13, 13, 13, 13, 13, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12)
start rechunking with (130, 122, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 48)
end of y_rechunk
CPU times: user 15.4 s, sys: 3.04 s, total: 18.4 s
Wall time: 1min 20s

<xarray.Dataset>
Dimensions:   (t: 31, x: 6560, y: 6540, z: 150)
Coordinates:
  * t         (t) object 2012-01-01 12:00:00 ... 2012-01-31 12:00:00
  * y         (y) int64 1 2 3 4 5 6 7 8 ... 6534 6535 6536 6537 6538 6539 6540
  * x         (x) int64 1 2 3 4 5 6 7 8 ... 6554 6555 6556 6557 6558 6559 6560
  * z         (z) int64 1 2 3 4 5 6 7 8 9 ... 143 144 145 146 147 148 149 150
    mask2d    (y, x) bool dask.array<chunksize=(130, 6560), meta=np.ndarray>
    nav_lon   (y, x) float32 dask.array<chunksize=(130, 6560), meta=np.ndarray>
    depth     (z, y, x) float32 dask.array<chunksize=(150, 130, 6560), meta=np.ndarray>
    mask      (z, y, x) bool dask.array<chunksize=(150, 130, 6560), meta=np.ndarray>
    nav_lat   (y, x) float32 dask.array<chunksize=(130, 6560), meta=np.ndarray>
Data variables:
    vosaline  (t, z, y, x) float32 dask.array<chunksize=(31, 150, 130, 6560), meta=np.ndarray>
    votemper  (t, z, y, x) float32 dask.array<chunksize=(31, 150, 130, 6560), meta=np.ndarray>
    vomecrty  (t, z, y, x) float32 dask.array<chunksize=(31, 150, 130, 6560), meta=np.ndarray>

%%time
monitor.auto(df,data,savefig,daskreport,outputpath,file_exp='SEDNA'
            )

#calc= True
#save= True
#plot= False
Zone='FramS'
Value='Section'
cmap='None'
clabel='None'
clim= {'vosaline': (33, 36.2), 'votemper': (-2, 6), 'vomecrty': (-0.05, 0.05)}
outputpath='../results/SEDNA_DELTA_MONITOR/'
nc_outputpath='../nc_results/SEDNA_DELTA_MONITOR/'
filename='SEDNA_section_FramS_Section'
#2 Zooming Data
dataa= zoom.FramS(data)

<xarray.Dataset>
Dimensions:   (t: 31, x: 554, z: 103)
Coordinates:
  * t         (t) object 2012-01-01 12:00:00 ... 2012-01-31 12:00:00
    y         int64 2609
  * x         (x) int64 3749 3750 3751 3752 3753 ... 4298 4299 4300 4301 4302
  * z         (z) int64 1 2 3 4 5 6 7 8 9 10 ... 95 96 97 98 99 100 101 102 103
    mask2d    (x) bool dask.array<chunksize=(554,), meta=np.ndarray>
    nav_lon   (x) float32 dask.array<chunksize=(554,), meta=np.ndarray>
    depth     (z, x) float32 dask.array<chunksize=(103, 554), meta=np.ndarray>
    mask      (z, x) bool dask.array<chunksize=(103, 554), meta=np.ndarray>
    nav_lat   (x) float32 dask.array<chunksize=(554,), meta=np.ndarray>
Data variables:
    vosaline  (t, z, x) float32 dask.array<chunksize=(31, 103, 554), meta=np.ndarray>
    votemper  (t, z, x) float32 dask.array<chunksize=(31, 103, 554), meta=np.ndarray>
    vomecrty  (t, z, x) float32 dask.array<chunksize=(31, 103, 554), meta=np.ndarray>

#3 Start computing 
dtaa= data.unify_chunks().persist()

<xarray.Dataset>
Dimensions:   (t: 31, x: 554, z: 103)
Coordinates:
  * t         (t) object 2012-01-01 12:00:00 ... 2012-01-31 12:00:00
    y         int64 2609
  * x         (x) int64 3749 3750 3751 3752 3753 ... 4298 4299 4300 4301 4302
  * z         (z) int64 1 2 3 4 5 6 7 8 9 10 ... 95 96 97 98 99 100 101 102 103
    mask2d    (x) bool dask.array<chunksize=(554,), meta=np.ndarray>
    nav_lon   (x) float32 dask.array<chunksize=(554,), meta=np.ndarray>
    depth     (z, x) float32 dask.array<chunksize=(103, 554), meta=np.ndarray>
    mask      (z, x) bool dask.array<chunksize=(103, 554), meta=np.ndarray>
    nav_lat   (x) float32 dask.array<chunksize=(554,), meta=np.ndarray>
Data variables:
    vosaline  (t, z, x) float32 dask.array<chunksize=(31, 103, 554), meta=np.ndarray>
    votemper  (t, z, x) float32 dask.array<chunksize=(31, 103, 554), meta=np.ndarray>
    vomecrty  (t, z, x) float32 dask.array<chunksize=(31, 103, 554), meta=np.ndarray>

#4 Saving  SEDNA_section_FramS_Section
dtaa=save.datas(data,plot=Plot,path=nc_outputpath,filename=filename)
start saving data
saving data in a file
t (31,)
0
slice(0, 31, None)

distributed.nanny - WARNING - Worker exceeded 95% memory budget. Restarting
distributed.nanny - WARNING - Restarting worker
distributed.nanny - WARNING - Worker exceeded 95% memory budget. Restarting
distributed.nanny - WARNING - Worker exceeded 95% memory budget. Restarting
distributed.nanny - WARNING - Restarting worker
distributed.nanny - WARNING - Restarting worker
distributed.nanny - WARNING - Worker exceeded 95% memory budget. Restarting
distributed.nanny - WARNING - Worker exceeded 95% memory budget. Restarting
distributed.nanny - WARNING - Restarting worker
distributed.nanny - WARNING - Restarting worker
distributed.nanny - WARNING - Worker exceeded 95% memory budget. Restarting
distributed.nanny - WARNING - Restarting worker
distributed.nanny - WARNING - Worker exceeded 95% memory budget. Restarting
distributed.nanny - WARNING - Restarting worker
distributed.nanny - WARNING - Worker exceeded 95% memory budget. Restarting
distributed.nanny - WARNING - Restarting worker
distributed.nanny - WARNING - Worker exceeded 95% memory budget. Restarting
distributed.nanny - WARNING - Restarting worker
distributed.nanny - WARNING - Worker exceeded 95% memory budget. Restarting
distributed.nanny - WARNING - Restarting worker
distributed.nanny - WARNING - Worker exceeded 95% memory budget. Restarting
distributed.nanny - WARNING - Restarting worker
distributed.nanny - WARNING - Worker exceeded 95% memory budget. Restarting
distributed.nanny - WARNING - Restarting worker
distributed.nanny - WARNING - Worker exceeded 95% memory budget. Restarting
distributed.nanny - WARNING - Restarting worker

---------------------------------------------------------------------------
KilledWorker                              Traceback (most recent call last)
<timed eval> in <module>

/ccc/work/cont003/gen7420/odakatin/monitor-sedna/notebook/core/monitor.py in auto(df, val, savefig, daskreport, outputpath, file_exp)
     77             print('dtaa=save.datas(data,plot=Plot,path=nc_outputpath,filename=filename)' )
     78             with performance_report(filename=daskreport+"_save_"+step.Value+".html"):
---> 79                 save.datas(data,plot=Plot,path=nc_outputpath,filename=filename)
     80     # 5. Plot
     81         if plotswitch=='True':

/ccc/work/cont003/gen7420/odakatin/monitor-sedna/notebook/core/save.py in datas(data, plot, path, filename)
     20         print('save computed data at',savedfile,'completed')
     21     elif 'section' in plot:
---> 22         twoD(data,path,filename,nested=False)
     23     else :
     24         twoD(data,path,filename)

/ccc/work/cont003/gen7420/odakatin/monitor-sedna/notebook/core/save.py in twoD(data, path, filename, nested)
     55     print('saving data in a file')
     56     filesave=path+filename
---> 57     return to_mfnetcdf_map(data,prefix=filesave, nested=nested)
     58 
     59 def twoD_onefile(data 

/ccc/work/cont003/gen7420/odakatin/monitor-sedna/notebook/core/save.py in to_mfnetcdf_map(ds, prefix, nested)
    226             ,template=template
    227             )  
--> 228         mapped.compute()
    229 
    230     return mapped#.compute()

~/monitor/lib/python3.7/site-packages/xarray/core/dataset.py in compute(self, **kwargs)
    904         """
    905         new = self.copy(deep=False)
--> 906         return new.load(**kwargs)
    907 
    908     def _persist_inplace(self, **kwargs) -> "Dataset":

~/monitor/lib/python3.7/site-packages/xarray/core/dataset.py in load(self, **kwargs)
    739 
    740             # evaluate all the dask arrays simultaneously
--> 741             evaluated_data = da.compute(*lazy_data.values(), **kwargs)
    742 
    743             for k, data in zip(lazy_data, evaluated_data):

~/monitor/lib/python3.7/site-packages/dask/base.py in compute(*args, **kwargs)
    561         postcomputes.append(x.__dask_postcompute__())
    562 
--> 563     results = schedule(dsk, keys, **kwargs)
    564     return repack([f(r, *a) for r, (f, a) in zip(results, postcomputes)])
    565 

~/monitor/lib/python3.7/site-packages/distributed/client.py in get(self, dsk, keys, workers, allow_other_workers, resources, sync, asynchronous, direct, retries, priority, fifo_timeout, actors, **kwargs)
   2653                     should_rejoin = False
   2654             try:
-> 2655                 results = self.gather(packed, asynchronous=asynchronous, direct=direct)
   2656             finally:
   2657                 for f in futures.values():

~/monitor/lib/python3.7/site-packages/distributed/client.py in gather(self, futures, errors, direct, asynchronous)
   1968                 direct=direct,
   1969                 local_worker=local_worker,
-> 1970                 asynchronous=asynchronous,
   1971             )
   1972 

~/monitor/lib/python3.7/site-packages/distributed/client.py in sync(self, func, asynchronous, callback_timeout, *args, **kwargs)
    837         else:
    838             return sync(
--> 839                 self.loop, func, *args, callback_timeout=callback_timeout, **kwargs
    840             )
    841 

~/monitor/lib/python3.7/site-packages/distributed/utils.py in sync(loop, func, callback_timeout, *args, **kwargs)
    338     if error[0]:
    339         typ, exc, tb = error[0]
--> 340         raise exc.with_traceback(tb)
    341     else:
    342         return result[0]

~/monitor/lib/python3.7/site-packages/distributed/utils.py in f()
    322             if callback_timeout is not None:
    323                 future = asyncio.wait_for(future, callback_timeout)
--> 324             result[0] = yield future
    325         except Exception as exc:
    326             error[0] = sys.exc_info()

~/monitor/lib/python3.7/site-packages/tornado/gen.py in run(self)
    760 
    761                     try:
--> 762                         value = future.result()
    763                     except Exception:
    764                         exc_info = sys.exc_info()

~/monitor/lib/python3.7/site-packages/distributed/client.py in _gather(self, futures, errors, direct, local_worker)
   1827                             exc = CancelledError(key)
   1828                         else:
-> 1829                             raise exception.with_traceback(traceback)
   1830                         raise exc
   1831                     if errors == "skip":

KilledWorker: ("('copy-rechunk-merge-getitem-075b14ae9e58a6ad4cece4baa71ae22b', 0, 0, 0)", <Worker 'tcp://127.0.0.1:35085', name: 4, memory: 0, processing: 1>)